gyli

# Lazy dict allows storing function and argument pairs when initializing the dictionary, 
# it calculates the value only when fetching it.
# In this examole, if the key starts with '#', it would accept a (function, args) tuple as value and 
# returns the calculated result when fetching the values.

from collections.abc import Mapping 


class LazyDict(Mapping):
    def __init__(self, *args, **kw):

gyli

def division(x: int, y: int, method: str = 'float') -> Union[int, float]:
    """
    :param x: dividend
    :param y: divisor
    :param method: {'float', 'round', 'ceil', 'floor'}
    """
    method_mapping = {
        'float': lambda a, b: a / b,
        'nearest': lambda a, b: round(a / b),
        'ceil': lambda a, b: int(a // b + bool(a % b)),

gyli

# This script generates a .py file with a dictionary of Font Awesome icon mapping, like '500px': '\uf26e'
# Run:
# python generate_fontawesome_mapping.py > fontawesome_mapping.py

import yaml
import requests
import sys


INDENT = ' ' * 4

gyli

# In CPython implementation of Python 3.6, dictionary keeps the insertion order.
# From Python 3.7, this will become a language feature.
# In order to sort a dictionary by key including nested dictionary inside, we can do:

def sort_dict(item: dict):
    """
    Sort nested dict
    Example:
         Input: {'a': 1, 'c': 3, 'b': {'b2': 2, 'b1': 1}}
         Output: {'a': 1, 'b': {'b1': 1, 'b2': 2}, 'c': 3}

gyli

import aiohttp
import asyncio


NUMBERS = range(12)
URL = 'http://httpbin.org/get?a={}'


async def fetch_async(a):
    async with aiohttp.ClientSession() as session:

gyli

import hashlib


def dict_hash(d: dict):
    """
    Hash dictionary's key and value in string as the hash value of the dict
    """
    out = hashlib.md5()
    for key, value in d.items():
        out.update(str(key).encode('utf-8'))

gyli

from itertools import islice
from typing import List, Iterator


def chunks(iterator: Iterator, step: int) -> List:
    """
    Yield successive step-sized chunks from iterator
    """
    iterable = iter(iterable)
    while True:

gyli

# Code is modified from Papertrail's official examples:
# https://help.papertrailapp.com/kb/configuration/configuring-centralized-logging-from-python-apps/#examples

import logging
import logging.config
import os
import socket

class ContextFilter(logging.Filter):
    """

gyli

def join_nullable(l: List[str], sep: str) -> str:
    """
    Filter out None in a list and join the remaining strings.
    Example:
        Input: l=['a', None, 'c'], sep=' '
        Output: 'a c'
    :param l: list of strings
    :param sep: separator
    """
    return sep.join(filter(lambda i: bool(i), l))

gyli

import pandas as pd
import multiprocessing as mp
import traceback


def processing_chunk(chunk):
    for row in chunk.iterrows():
        pass
    time.sleep(5)