ORIONIS API REFERENCE

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

orionis.queues.serializer.JobSerializer

class documentation

class JobSerializer(IJobSerializer): (source)

Constructor: JobSerializer()

View In Hierarchy

Serialize explicit job state through a registry of application-trusted types.

Method __init__ Initialize reusable codecs and an empty trusted registry.
Method _checkSize Reject oversized or nonbinary wire data before decoding.
Method _validateFields Validate annotated fields without accepting arbitrary application types.
Method decode Restore primitive fields for a registered identity without running init.
Method decodeEnvelope Validate untrusted envelope JSON without resolving its job type.
Method encode Serialize declared fields while rejecting service and undeclared state.
Method encodeEnvelope Encode the immutable envelope as explicit versioned JSON.
Method register Register a concrete async job class supplied by trusted application code.
Class Variable __slots__ Undocumented
Instance Variable _decoder Undocumented
Instance Variable _encoder Undocumented
Instance Variable _registry Undocumented
Instance Variable _state_decoder Undocumented
Instance Variable _types Undocumented
def __init__(self): (source)

Initialize reusable codecs and an empty trusted registry.

Returns
NoneRetain codecs without importing application job modules.
def _checkSize(self, payload: bytes): (source)

Reject oversized or nonbinary wire data before decoding.

Parameters
payload:bytesSerialized job or envelope.
Returns
NoneEnforce the wire-size limit.
def _validateFields(self, metadata: _JobMetadata, values: dict[str, object]): (source)

Validate annotated fields without accepting arbitrary application types.

Parameters
metadata:_JobMetadataTrusted persistent field constraints.
values:dict[str, object]Restored or declared primitive state.
Returns
NoneValidate field types without invoking job constructors.
def decode(self, identity: str, payload: bytes) -> BaseJob: (source)
overrides orionis.queues.contracts.serializer.IJobSerializer.decode

Restore primitive fields for a registered identity without running init.

Parameters
identity:strStable trusted identity, never an instruction to import a module.
payload:bytesUntrusted MessagePack state.
Returns
BaseJobReconstructed job awaiting scoped DI invocation.
def decodeEnvelope(self, payload: bytes) -> JobEnvelope: (source)
overrides orionis.queues.contracts.serializer.IJobSerializer.decodeEnvelope

Validate untrusted envelope JSON without resolving its job type.

Parameters
payload:bytesStored wire data.
Returns
JobEnvelopeImmutable options ready for registry-based state decoding.
def encode(self, job: BaseJob) -> tuple[str, bytes]: (source)
overrides orionis.queues.contracts.serializer.IJobSerializer.encode

Serialize declared fields while rejecting service and undeclared state.

Parameters
job:BaseJobTrusted application job instance.
Returns
tuple[str, bytes]Stable identity and primitive MessagePack payload.
Raises
QueuePayloadErrorIf state is undeclared, unsupported, or exceeds the payload limit.
def encodeEnvelope(self, envelope: JobEnvelope) -> bytes: (source)
overrides orionis.queues.contracts.serializer.IJobSerializer.encodeEnvelope

Encode the immutable envelope as explicit versioned JSON.

Parameters
envelope:JobEnvelopeValidated execution data.
Returns
bytesJSON suitable for native binary database or Redis storage.
def register(self, job_type: type[BaseJob]): (source)
overrides orionis.queues.contracts.serializer.IJobSerializer.register

Register a concrete async job class supplied by trusted application code.

Parameters
job_type:type[BaseJob]Class imported by application bootstrap or dispatched locally.
Returns
NoneCache metadata once for the stable identity.
__slots__: tuple[str, ...] = (source)
overrides orionis.queues.contracts.serializer.IJobSerializer.__slots__

Undocumented

_decoder = (source)

Undocumented

_encoder = (source)

Undocumented

_registry: dict[str, _JobMetadata] = (source)

Undocumented

_state_decoder = (source)

Undocumented

_types: dict[type[BaseJob], tuple[str, _JobMetadata]] = (source)

Undocumented