chancy.utils module

class chancy.utils.TaskManager

Bases: object

A simple task manager that keeps track of tasks.

add(task: Coroutine, *, name: str | None = None) → Task

Add a task to the manager.

async cancel_all()

Cancel all tasks and wait for them to complete.

async wait_for_shutdown()

Wait until all tasks are complete, including any that are added after this method is called.

chancy.utils.chancy_uuid() → str

Generate a UUID suitable for use as a job ID.

Note

It’s UUID7, kinda, since the draft keeps changing.

Returns:

str

chancy.utils.chunked(iterable, size)

Yield chunks of size from iterable.

Parameters:

• iterable – The iterable to chunk.

• size – The size of each chunk.

chancy.utils.import_string(name)

Import an object from a string previously created using importable_name().

Parameters:

name – The importable name of the object.

Returns:

Any

chancy.utils.importable_name(obj)

Get the importable name for an object.

Note

This will only work for objects that are actually importable, i.e. they are defined in a module. Lambdas, for example, will not have an importable name.

Parameters:

obj – The object to get the importable name for.

Returns:

str

chancy.utils.json_dumps(obj, **kwargs)

Serialize an object to a JSON formatted str with support for UUIDs.

Parameters:

• obj – The object to serialize.

• kwargs – Additional arguments to pass to json.dumps.

Returns:

str

async chancy.utils.sleep(seconds: int, *, events: Iterable[Coroutine] | None = None) → bool

Sleep for a specified number of seconds, or until one of the given events occurs.

chancy.utils.timed_block()

A context manager that times the block of code within it. The elapsed time will be updated in the elapsed attribute of the returned object until the block is exited.

Example:

with timed_block() as timer:
    time.sleep(1)

print(f"Block took {timer.elapsed} seconds.")