Skip to content

Instantly share code, notes, and snippets.

SELECT table,
formatReadableSize(sum(bytes)) as size,
min(min_date) as min_date,
max(max_date) as max_date
FROM system.parts
WHERE active
GROUP BY table
@elessarelfstone
elessarelfstone / luigi_ftp_complete_override.py
Created August 26, 2023 11:22
Luigi task - remove file before uploading
import luigi
from luigi.contrib import ftp
class UploadFileToFTP(luigi.Task):
file_path = luigi.Parameter()
delete_file = luigi.BoolParameter(default=False)
def output(self):
return ftp.RemoteTarget('ftp.example.com', self.file_path)
@elessarelfstone
elessarelfstone / wsl-cheatsheet.ps1
Created July 25, 2022 05:50 — forked from karthiks/wsl-cheatsheet.ps1
WSL 2 CLI Cheat-sheet To Be Run In Powershell
# To list installed distributions
wsl -l
wsl --list
# To list installed distributions along with its running status and wsl config being 1 or 2
wsl -l --verbose
wsl -l -v
# To run a specific distro
wsl -d distro_name
# change to postgres user and open psql prompt
sudo -u postgres psql postgres
# list databases
postgres=# \l
# list roles
postgres=# \du
# create role
@elessarelfstone
elessarelfstone / luigi_time_tasks_example.py
Created December 13, 2020 12:18 — forked from samuell/luigi_time_tasks_example.py
How to output the execution time of tasks in the luigi workflow system, as discussed [here](https://groups.google.com/d/msg/luigi-user/uivbf-luX9w/z0GCKKsIefoJ)
import luigi
import time
class TimeTaskMixin(object):
'''
A mixin that when added to a luigi task, will print out
the tasks execution time to standard out, when the task is
finished
'''
@luigi.Task.event_handler(luigi.Event.PROCESSING_TIME)
@elessarelfstone
elessarelfstone / python_simple_threading.py
Created November 14, 2020 14:43
python_simple_threading
def sync_events():
ts = get_last_event_ts()
if not ts:
ts = datetime.now()
while True:
ts = add_new_events_to_notify(ts)
if not ts:
ts = get_last_event_ts()
sleep(DB_PROCESS_TIMEOUT)
@elessarelfstone
elessarelfstone / sqlalchemy.py
Last active November 13, 2020 18:32
sqlalchemy
class FailedEvents(Base):
__tablename__ = 'failed_events'
event_id = Column(Integer, primary_key=True)
delivered = Column(Boolean)
def get_last_event_ts() -> datetime:
session = Session()
max_ts = session.query(func.max(TasksEvents.ts)).\
select_from(TasksEvents, FailedEvents).\
filter(TasksEvents.id == FailedEvents.event_id).\
@elessarelfstone
elessarelfstone / python_dbapi.py
Last active November 13, 2020 17:28
sample loading from sqlite
def make_conn():
return sqlite3.connect(LUIGI_HISTORY_DBPATH)
def load_new_luigi_events(time: str) -> List[Dict]:
""" Load luigi events which happened after some time point"""
cursor = make_conn().cursor()
cursor.execute(SQL, [time])
cols = [col[0] for col in cursor.description]
return [dict(zip(cols, row)) for row in cursor.fetchall()]
@elessarelfstone
elessarelfstone / django_cheat_sheet.md
Created September 9, 2019 05:06 — forked from bradtraversy/django_cheat_sheet.md
Django command cheat sheet

Django 2.x Cheat Sheet

Creating a virtual environment

We need to create a virtual env for our app to run in: More Here Run this command in whatever folder you want to create your venv folder

python -m venv ./venv
@elessarelfstone
elessarelfstone / django_deploy.md
Created September 9, 2019 05:04 — forked from bradtraversy/django_deploy.md
Django Deployment - Digital Ocean

Django Deployment to Ubuntu 18.04

In this guide I will go through all the steps to create a VPS, secure it and deploy a Django application. This is a summarized document from this digital ocean doc

Any commands with "$" at the beginning run on your local machine and any "#" run when logged into the server

Create A Digital Ocean Droplet

Use this link and get $10 free. Just select the $5 plan unless this a production app.