utils

Provides utility functions for use throughout the code.

ClassRegistry

Bases: ABC, Generic[T]

A registry of Plugboard classes.

add classmethod

add(
    plugboard_class: type[T], key: Optional[Hashable] = None
) -> None

Add a class to the registry.

Parameters:

Name	Type	Description	Default
plugboard_class	type[T]	The class to register.	required
key	Optional[Hashable]	Optional; The key to register the class under.	None

build classmethod

build(
    plugboard_class: Hashable, *args: Any, **kwargs: Any
) -> T

Builds a Plugboard object.

Parameters:

Name	Type	Description	Default
plugboard_class	Hashable	The key corresponding to the required class.	required
*args	Any	Positional arguments to pass to the class constructor.	()
**kwargs	Any	Keyword arguments to pass to the class constructor.	{}

Returns:

Type	Description
T	An object of the required class.

get classmethod

get(plugboard_class: Hashable) -> type[T]

Returns a class from the registry.

Parameters:

Name	Type	Description	Default
plugboard_class	Hashable	The key corresponding to the required class.	required

Returns:

Type	Description
type[T]	The class.

DI

Bases: BaseContainer

DI is a dependency injection container for plugboard.

EntityIdGen

EntityIdGen generates entity ids.

id classmethod

id(entity: Entity) -> str

Returns a unique entity id.

Parameters:

Name	Type	Description	Default
entity	Entity	The entity to generate an id for.	required

Returns:

Name	Type	Description
str	str	The generated id.

is_job_id classmethod

is_job_id(id: str) -> bool

Checks if an id is a job id.

Parameters:

Name	Type	Description	Default
id	str	The id to check.	required

Returns:

Name	Type	Description
bool	bool	True if the id is a job id.

job_id classmethod

job_id() -> str

Returns a unique job id.

Returns:

Name	Type	Description
str	str	The generated job id.

parse classmethod

parse(id: str) -> tuple[Entity, str]

Parses an entity id.

Parameters:

Name	Type	Description	Default
id	str	The entity id to parse.	required

Returns:

Type	Description
tuple[Entity, str]	tuple[Entity, str]: The parsed entity and id.

ExportMixin

AsDictMixin provides functionality for converting objects to dict.

_convert_exportable_objs staticmethod

_convert_exportable_objs(obj: Any) -> _t.Any

Recursively converts Exportable objects to their export representation.

dict

dict() -> dict

Returns dict representation of object.

export

export() -> dict

Returns dict representation of object for later reconstruction.

json

json() -> bytes

Returns JSON representation of object as bytes.

Exportable

Bases: Protocol

Exportable protocol for objects that can be exported.

export

export() -> dict

Returns dict representation of object for later reconstruction.

add_sys_path

add_sys_path(path: str | PathLike) -> Iterator

Temporarily add path to sys.path.

build_actor_wrapper

build_actor_wrapper(cls: type[T]) -> type[_ActorWrapper[T]]

Builds an actor wrapper around a class.

This is useful for handling classes that are modified at runtime, e.g. via wrapped methods, and therefore not supported by the ray.remote decorator.

The wrapper methods will have the same name as the original methods, but where nested in class attributes the method names will be prefixed accordingly. The wrapper also provides a getattr and setattr method to access the wrapped object's properties.

Parameters:

Name	Type	Description	Default
cls	type[T]	The class to wrap.	required

Returns:

Type	Description
type[_ActorWrapper[T]]	A new class that wraps the original class and can be used as a Ray actor.

depends_on_optional

depends_on_optional(
    module_name: str, extra: Optional[str] = None
) -> _t.Callable

Decorator to check for optional dependencies.

Parameters:

Name	Type	Description	Default
module_name	str	The name of the module to check for.	required
extra	Optional[str]	Optional; The name of the extra that the module is associated with. Defaults to the module name.	None

gather_except async

gather_except(*coros: Coroutine) -> list[_t.Any]

Attempts to gather the given coroutines, raising any exceptions.

gen_rand_str

gen_rand_str(chars: int = RANDOM_CHAR_COUNT) -> str

Generates a random string of a fixed length and character set.

With 16 chars in [a-zA-Z0-9], at 1000 ids/second it would take ~1000 years or 30T ids for >= 1% chance of at least one collision. See here for details: https://zelark.github.io/nano-id-cc/

Note: This function is not suitable for cryptographic purposes; it is intended to generate random strings for unique identifiers only.

Returns:

Name	Type	Description
str	str	Random fixed length string.

is_on_ray_worker

is_on_ray_worker() -> bool

Returns True if called from a Ray worker.