chancy.utils module¶

class chancy.utils.DatabaseConnection[source]¶

Bases: TypedDict

Django-style database connection configuration.

ENGINE: str[source]¶

HOST: NotRequired[str][source]¶

NAME: NotRequired[str][source]¶

OPTIONS: NotRequired[dict[str, Any]][source]¶

PASSWORD: NotRequired[str][source]¶

PORT: NotRequired[str][source]¶

USER: NotRequired[str][source]¶

class chancy.utils.TaskManager[source]¶

Bases: object

A simple task manager that keeps track of tasks.

add(name, task)[source]¶

Add a task to the manager.

Parameters:

name (str)
task (Coroutine)

Return type:

Task

async cancel(name)[source]¶

Cancel a task by name and wait for it to complete.

If the task is not found, a KeyError is raised.

Parameters:: name (str)

async cancel_all()[source]¶: Cancel all tasks and wait for them to complete.

async wait_for_shutdown()[source]¶: Wait until all tasks are complete, including any that are added after this method is called.

chancy.utils.chancy_uuid()[source]¶

Generate a UUID suitable for use as a job ID.

Note

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

Returns:: UUID
Return type:: UUID

chancy.utils.chunked(iterable, size)[source]¶

Yield chunks of size from iterable.

Parameters:

iterable – The iterable to chunk.
size – The size of each chunk.

chancy.utils.get_database_dsn(db_settings)[source]¶

Convert a django-style database connection configuration to a DSN.

Parameters:: db_settings (DatabaseConnection)
Return type:: str

chancy.utils.import_string(name)[source]¶

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)[source]¶

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)[source]¶

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

chancy.utils.lock_order_key(unique_key)[source]¶

The key used to order job rows before locking them.

Every code path that takes row locks on several jobs in one transaction (pushing jobs with unique keys, the worker’s batched job updates) must sort by this key so that overlapping transactions acquire their locks in the same order and cannot deadlock. See issue #89.

Parameters:: unique_key (str | None)
Return type:: str

async chancy.utils.sleep(seconds, *, events=None)[source]¶

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

Parameters:

seconds (int)
events (Iterable[Coroutine] | None)

Return type:

bool

chancy.utils.timed_block()[source]¶

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.")