brettvitaz

def hash_list(list_: list) -> int:
    __hash = 0
    for i, e in enumerate(list_):
        __hash = hash((__hash, i, hash_item(e)))
    return __hash


def hash_dict(dict_: dict) -> int:
    __hash = 0
    for k, v in dict_.items():

brettvitaz

from abc import ABC


class JsonSerializable(ABC):
    IGNORE = []

    @staticmethod
    def serialize(obj):
        if isinstance(obj, JsonSerializable):
            return obj.to_json()

brettvitaz

from collections import Counter
from dataclasses import dataclass, field

from dataclasses_json import dataclass_json

from utils import counter_config


@dataclass_json
@dataclass

brettvitaz

from typing import Union

import json


class AppSettings(dict):
    def __init__(self, settings_filename):
        self._settings_filename = settings_filename
        with open(settings_filename, "r") as settings_file:
            super().__init__(json.load(settings_file))

brettvitaz

def prune_dict_for_dataclass(d: dict, cls):
    ret = {}
    for field in fields(cls):
        if field.name in d:
            if is_dataclass(field.type):
                ret[field.name] = prune_dict_for_dataclass(d[field.name], field.type)
            elif typing.get_origin(field.type) is list and type(d[field.name]) is list:
                list_cls = typing.get_args(field.type)[0]
                if is_dataclass(list_cls):
                    ret[field.name] = [prune_dict_for_dataclass(i, list_cls) for i in d[field.name]]

brettvitaz

def print_progress_bar(iteration, total, prefix='', suffix='', decimals=1, length=100, fill='█', print_end=""):
    percent = ("{0:." + str(decimals) + "f}").format(100 * (iteration / float(total)))
    filled_length = int(length * iteration // total)
    bar = fill * filled_length + '-' * (length - filled_length)
    print(f'\r{prefix} |{bar}| {percent}% {suffix}', end=print_end)
    # Print New Line on Complete
    if iteration == total:
        print()

brettvitaz

async def gather_with_concurrency(n, *coros):
    semaphore = asyncio.Semaphore(n)

    async def sem_coro(coro):
        async with semaphore:
            return await coro

    return await asyncio.gather(*(sem_coro(c) for c in coros))