Python Libraries Video 1

argh_examples.py

import argh

def do_the_thing(required_arg, optional_arg=1, other_optional_arg=False):
    """
    I am a docstring
    """
    print((required_arg, type(required_arg)))
    print((optional_arg, type(optional_arg)))
    print((other_optional_arg, type(other_optional_arg)))


@argh.arg('--bool-arg-for-flag', '-b', help="Flip this flag for things")
@argh.arg('arg_with_choices', choices=['one', 'two', 'three'])
def do_the_other_thing(arg_with_choices, bool_arg_for_flag=False):
    print(arg_with_choices)
    print(bool_arg_for_flag)


if __name__ == '__main__':
    # argh.dispatch_command(do_the_thing)
    argh.dispatch_commands([do_the_thing, do_the_other_thing])

msgpack_examples.py

import msgpack
import json
import random


def msgpack_example_1():
    example_dict = {i: random.random() for i in range(10000)}

    with open('json_file.json', 'w') as f:
        json.dump(example_dict, f)
    with open('json_file.json') as f:
        back_from_json = json.load(f)

    # Saving and loading
    with open('msgpack_file.msgpack', 'wb') as f:
        # f.write(msgpack.packb(example_dict))
        f.write(msgpack.packb(example_dict, use_single_float=True))
    with open('msgpack_file.msgpack', 'rb') as f:
        back_from_msgpack = msgpack.unpackb(f.read())

    # Data integrity
    print(type(next(iter(back_from_json.keys()))))
    print(type(next(iter(back_from_msgpack.keys()))))


def msgpack_example_2():
    list_of_dicts = [{0: random.random()} for i in range(100)]
    with open('streamed.msgpack', 'wb') as f:
        for d in list_of_dicts:
            f.write(msgpack.packb(d))

    with open('streamed.msgpack', 'rb') as f:
        loaded_list_of_dicts = [item for item in msgpack.Unpacker(f)]

    print(list_of_dicts[3][0], loaded_list_of_dicts[3][0])

if __name__ == '__main__':
    # msgpack_example_1()
    msgpack_example_2()

redis_cache_examples.py

# sudo apt install redis-server
# sudo systemctl enable redis-server.service
# sudo systemctl start redis-server.service

# pip/pip3 install git+https://github.com/YashSinha1996/redis-simple-cache.git

import time
from redis_cache import cache_it, cache_it_json


@cache_it(limit=1000, expire=5)
def function_that_takes_a_long_time(i):
    print(f"function was called with input {i}")
    return i**2


if __name__ == '__main__':
    for i in range(10):
        print(i, function_that_takes_a_long_time(2))

schedule_examples.py

import time
import schedule

def test_function():
    print(f'test called at {time.time()}')

def test_function_2():
    print(f'test 2 called at {time.time()}')

if __name__ == '__main__':
    schedule.every(1).seconds.do(test_function)
    schedule.every(3).seconds.do(test_function_2)
    # schedule.every(1).days.do(daily_task)
    # schedule.every().thursday.at("10:00").do(day_time_task)

    while True:
        schedule.run_pending()

tqdm_examples.py

from tqdm import tqdm, trange
import random
import time


def tqdm_example_1():
    for i in tqdm(range(10)):
        time.sleep(0.2)


def tqdm_example_2():
    for i in trange(10, desc="outer_loop"):
        for j in trange(10, desc="inner_loop"):
            time.sleep(0.01)


def tqdm_example_3(add_tot=False):
    max_iter = 100
    tot = 0

    if add_tot:
        bar = tqdm(desc="update example", total=max_iter)
    else:
        bar = tqdm()

    while tot < max_iter:
        update_iter = random.randint(1, 5)
        bar.update(update_iter)
        tot += update_iter
        time.sleep(0.03)


def tqdm_example_4():
    t = trange(100)
    for i in t:
        t.set_description(f"on iter {i}")
        time.sleep(0.02)


if __name__ == "__main__":
    # tqdm_example_1()
    # tqdm_example_2()
    # tqdm_example_3()
    # tqdm_example_3(True)
    tqdm_example_4()