Useful Utils

gd.py provides a plenty of useful and handy functions.

Async Helpers

Whether they are for async things:

import asyncio

def file_io():
    with open('some_file.log', 'w') as file:
        file.write('test')

await gd.utils.run_blocking_io(file_io)

await gd.utils.gather(
    asyncio.sleep(1),
    asyncio.sleep(2),
    asyncio.sleep(3)
)  # will sleep for 3 seconds only

await gd.utils.run_blocking_io(func: Callable, *args, **kwargs) → Any

This function is a coroutine.

Run some blocking function in an event loop.

If there is a running loop, 'func' is executed in it.

Otherwise, a new loop is being created and closed at the end of the execution.

Example:

def make_image():
    ...  # long code of creating an image

# somewhere in an async function:

await run_blocking_io(make_image)

gd.utils.run(coro: Coroutine, *, loop: Optional[asyncio.events.AbstractEventLoop] = None, debug: bool = False, set_to_none: bool = False) → Any

Run a coroutine.

This function runs the passed coroutine, taking care of the event loop and shutting down asynchronous generators.

This function is basically ported from Python 3.7 for backwards compability with earlier versions of Python.

This function cannot be called when another event loop is running in the same thread.

If debug is True, the event loop will be run in debug mode.

This function creates a new event loop and closes it at the end if a loop is None.

If a loop is given, this function basically calls asyncio.AbstractEventLoop.run_until_complete().

It should be used as a main entry point to asyncio programs, and should ideally be called only once.

Example:

async def test(pid):
    return pid

one = gd.utils.run(test(1))

Parameters

• coro (coroutine) – Coroutine to run.

• loop (Optional[asyncio.AbstractEventLoop]) – A loop to run coro with. If None or omitted, a new event loop is created.

• debug (bool) – Whether or not to run event loop in debug mode.

• set_to_none (bool) – Indicates if the loop should be set to None after execution.

Returns

Anything that coro returns.

Return type

Any

gd.utils.acquire_loop(running: bool = False) → asyncio.events.AbstractEventLoop

Gracefully acquire a loop.

The function tries to get an event loop via asyncio.get_event_loop(). On fail, returns a new loop using asyncio.new_event_loop().

Parameters

running (bool) – Indicates if the function should get a loop that is already running.

gd.utils.cancel_all_tasks(loop: asyncio.events.AbstractEventLoop) → None

Cancels all tasks in a loop.

Parameters

loop (asyncio.AbstractEventLoop) – Event loop to cancel tasks in.

await gd.utils.wait(*fs: Iterable[Coroutine], loop: Optional[asyncio.events.AbstractEventLoop] = None, timeout: Optional[Union[float, int]] = None, return_when: str = 'ALL_COMPLETED') → Tuple[Set[_asyncio.Future], Set[_asyncio.Future]]

A function that is calling asyncio.wait().

Used for less imports inside and outside of this library.

Wait for the Futures and coroutines given by fs to complete.

The sequence futures must not be empty.

Coroutines will be wrapped in Tasks.

Returns two sets of Future: (done, pending).

Usage:

done, pending = await gd.utils.wait(fs)

Note

This does not raise TimeoutError! Futures that aren’t done when the timeout occurs are returned in the second set.

await gd.utils.gather(*aws: Sequence[Awaitable], loop: Optional[asyncio.events.AbstractEventLoop] = None, return_exceptions: bool = False) → List[Any]

A function that is calling asyncio.gather().

Used for less imports inside and outside gd.py.

One small addition is that a sequence of awaitables can be given as the only positional argument.

This way, asyncio.gather() will be run on that sequence.

Sequence Helpers

Or for sequences:

client = gd.Client()

level = await client.get_level(30029017)

comments = await level.get_comments(amount=1000)

over_100_likes = gd.utils.find(lambda comment: comment.rating > 100, comments)
# may be None, though

gd.utils.find(predicate: Callable[[Any], bool], iterable: Iterable, *, find_all: bool = False) → Union[Any, List[Any]]

For each element in iterable, return first element if predicate returns True and 'find_all' is False.

Otherwise, find all elements matching and return them.

Example:

...
friends = await client.get_friends()
old_users = gd.utils.find(lambda x: x.account_id < 500000, friends, find_all=True)

gd.utils.get(iterable: Iterable, **attrs: Any) → Union[Any, List[Any]]

For each element in iterable, return first element that matches requirements and 'find_all' is False.

Otherwise, find all elements matching and return them.

Example:

...
friends = await client.get_friends()
nekit = gd.utils.get(friends, name='nekitdev')

gd.utils.unique(iterable: Iterable) → List[Any]

Return a list of all unique elements in iterable.

This function preserves order of elements.

Example:

unique([3, 2, 1, 1, 2]) -> [3, 2, 1]