utils

Complexity

Total Complexity

6

Size/Duplication

Total Lines	36
Duplicated Lines	0 %

Importance

Changes

0

3 Functions

Rating	Name	Duplication	Size	Complexity
A	get_digest()	0	2	1
A	batch()	0	8	3
A	event_wait()	0	6	2

import asyncio
import hashlib
import logging
import pickle

from itertools import chain, islice


_logger = logging.getLogger(__name__)


class ClientNotFound(Exception):
    pass


async def event_wait(evt, timeout) -> bool:
    try:
        await asyncio.wait_for(evt.wait(), timeout)
    except asyncio.TimeoutError:
        pass
    return evt.is_set()


def get_digest(conf) -> str:
    return hashlib.md5(pickle.dumps(conf)).hexdigest()


def batch(iterable, size):
    iterator = iter(iterable)
    while True:
        try:
            batchiter = islice(iterator, size)
            yield list(chain([next(batchiter)], batchiter))
        except StopIteration:
            break