orionis.queues.serializer

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

module documentation

(source)

Undocumented

Class	`_JobMetadata`	Retain trusted class metadata independently of live job instances.
Class	`JobSerializer`	Serialize explicit job state through a registry of application-trusted types.
Function	`_collect_job_fields`	Collect one job class's declared slots and persistent annotations.
Function	`_job_metadata`	Collect inherited persistent slots and annotations from a trusted class.
Function	`_pack_value`	Validate owned primitive data and tag tuples in MessagePack.
Function	`_unpack_extension`	Restore the tuple extension without importing or invoking payload types.
Function	`_validate_annotation`	Check whether a field annotation contains only supported state types.
Function	`_validate_unpacked_value`	Validate decoded primitive state without rebuilding its containers.
Constant	`_MAX_PAYLOAD_BYTES`	Undocumented
Constant	`_MAX_STATE_DEPTH`	Undocumented
Constant	`_OPTIONS`	Undocumented
Constant	`_SCALAR_TYPES`	Undocumented
Constant	`_TUPLE_EXTENSION`	Undocumented

def _collect_job_fields(parent: type[BaseJob], names: dict[str, None], annotations: dict[str, object]): (source) ¶

Collect one job class's declared slots and persistent annotations.

Parameters
parent:`type[BaseJob]`	Registered job class or one of its BaseJob ancestors.
names:`dict[str, None]`	Ordered persistent field names collected across the hierarchy.
annotations:`dict[str, object]`	Field constraints updated by more specific classes.

Returns
`None`	Update the field names and annotations in place.

def _job_metadata(job_type: type[BaseJob]) -> _JobMetadata: (source) ¶

Collect inherited persistent slots and annotations from a trusted class.

Parameters
job_type:`type[BaseJob]`	Concrete job class registered by application code.

Returns
`_JobMetadata`	Persistent field names and optional primitive type constraints.

Raises
`QueueConfigurationError`	If persistent fields are private, shadow job options, or use unsupported annotations.

def _pack_value(value: object, depth: int = 0) -> object: (source) ¶

Validate owned primitive data and tag tuples in MessagePack.

Parameters
value:`object`	Field value or nested item.
depth:`int, optional`	Current nesting level.

Returns
`object`	Data supported by the MessagePack encoder.

Raises
`QueuePayloadError`	If state exceeds the depth limit, contains unsupported objects, nonfinite numbers, or dictionary keys that are not plain strings.

def _unpack_extension(code: int, data: memoryview) -> tuple[object, ...]: (source) ¶

Restore the tuple extension without importing or invoking payload types.

Parameters
code:`int`	MessagePack extension discriminator.
data:`memoryview`	Encoded tuple items.

Returns
`tuple[object, ...]`	Restored primitive tuple.

def _validate_annotation(annotation: object) -> bool: (source) ¶

Check whether a field annotation contains only supported state types.

Parameters
annotation:`object`	Trusted class field annotation.

Returns
`bool`	Whether msgspec can validate this field without constructing services.

def _validate_unpacked_value(value: object, depth: int = 0): (source) ¶

Validate decoded primitive state without rebuilding its containers.

Parameters
value:`object`	Decoded field value or nested item.
depth:`int, optional`	Current nesting level.

Returns
`None`	Accept supported state without copying decoded containers.

Raises
`QueuePayloadError`	If state exceeds the depth limit or contains unsupported values.

_MAX_PAYLOAD_BYTES = (source) ¶

Undocumented

Value

1024 * 1024

_MAX_STATE_DEPTH: int = (source) ¶

Undocumented

Value

_OPTIONS = (source) ¶

Undocumented

Value

frozenset(set(['tries', 'timeout', 'backoff']))

_SCALAR_TYPES = (source) ¶

Undocumented

Value

frozenset(set([str, bytes, int, float, bool, type(None)]))

_TUPLE_EXTENSION: int = (source) ¶

Undocumented

Value