kgantsov

from datetime import datetime
from datetime import timedelta
from uuid import UUID


def time_from_uuid1(u: str) -> datetime:
    return datetime(1582, 10, 15) + timedelta(microseconds=UUID(u).time // 10)

kgantsov

fastapi==0.70.0
gino==1.0.1
pytest==6.2.5
pytest-asyncio==0.16.0
requests==2.26.0

kgantsov

# simple script that finds all python virtualenv folders and shows how much
# space they are taking byt running `du -sh` command on each of the found directory
sudo find ~ -type d \( -name env -o -name venv \) -exec du -sh {} \; 2>/dev/null

kgantsov

- What do Etcd, Consul, and Zookeeper do?
  - Service Registration:
    - Host, port number, and sometimes authentication credentials, protocols, versions
      numbers, and/or environment details.
  - Service Discovery:
    - Ability for client application to query the central registry to learn of service location.
  - Consistent and durable general-purpose K/V store across distributed system.
    - Some solutions support this better than others.
    - Based on Paxos or some derivative (i.e. Raft) algorithm to quickly converge to a consistent state.
    - Centralized locking can be based on this K/V store.

kgantsov

covid-19-dashboard-styles

A Pen by Konstantin on CodePen.

License.

kgantsov

# Usage: echo '{"a": [], "b": "c"}' | jsonify
alias jsonify="python -m json.tool | pygmentize -l json"

kgantsov

# Usage: whoseport 8888
whoseport () {
 lsof -i ":$1" | grep --color=auto --exclude-dir={.bzr,CVS,.git,.hg,.svn} LISTEN
}

kgantsov

#!/usr/bin/python
import os

import usb

import telebot
from temperusb import TemperDevice

TEMP_VENDOR_ID = 0x0c45
TEMP_PRODUCT_ID = 0x7401

kgantsov

from celery.task.control import revoke
from celery.task.control import inspect

def revoke_tasks_by_name(task_name, worker_prefix=''):
    """
    Revoke all tasks by the name of the celery task

    :param task_name: Name of the celery task
    :param worker_prefix: Prefix for the worker

kgantsov

from datetime import datetime
import time

import kombu.five
from celery.task.control import inspect
    
def get_stuck_celery_tasks(threshold=600):
    i = inspect()

    for worker, tasks in i.active().items():