kissgyorgy

#!/bin/bash
function f() {
	sleep "$1"
    echo "$1"
}
while [ -n "$1" ]
do
    f $1 &
	shift
done

kissgyorgy

class LazyField:
    """Evaluates __repr__ lazily for heavy calculations.
    Expects callable with args and kwargs of how to calculate the value.
    """

    def __init__(self, func, *args, **kwargs):
        self._func = func
        self._args = args
        self._kwargs = kwargs
        self._value = ...

kissgyorgy

#!/bin/bash

# Wait for a socket on given host and port to became active.
# Will exit with statuscode 0 if succeeded, -1 otherwise.
# Originally got it from the blog post: https://developer.atlassian.com/blog/2015/03/docker-systemd-socket-activation/

host=$1
port=$2
tries=600

kissgyorgy

import asyncio
import httpx


async def infinite_loop(backend_client):
    while True:
        try:
            await asyncio.sleep(3)
        except asyncio.CancelledError:
            print("Cancelled")

kissgyorgy

import requests
import datetime

URI = 'https://gist.githubusercontent.com/jorin-vogel/7f19ce95a9a842956358/raw/e319340c2f6691f9cc8d8cc57ed532b5093e3619/data.json'
filename = datetime.date.today().strftime('%Y%m%d.csv')


def imperative():
    with open(filename, 'w') as f:
        f.write('Name,Credit Card\n')

kissgyorgy

# https://www.python.org/dev/peps/pep-0525/
import asyncio


async def generate_numbers():
    i = 0
    while True:
        try:
            yield i
        except GeneratorExit:

kissgyorgy

# For example, if the input is 5, the shape should look like this:
#   *
#  ***
# *****
#  ***
#   *
import math

lines = int(input('How many lines? '))

kissgyorgy

def _calculate_hash_and_save(uploaded_file, temp_file):
    file_hash = hashlib.sha256()
    for chunk in uploaded_file.chunks():
        file_hash.update(chunk)
        temp_file.write(chunk)
    return file_hash.hexdigest()

kissgyorgy

from concurrent.futures import ProcessPoolExecutor, wait

WORKERS = 6
futures = []


def cpu_hog():
    print("Starting computation")
    return 2 ** 10000000000000000

kissgyorgy

import asyncio
import psycopg2

# dbname should be the same for the notifying process
conn = psycopg2.connect(host="localhost", dbname="example", user="example", password="example")
conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

cursor = conn.cursor()
cursor.execute(f"LISTEN match_updates;")