temporalio.converter.ExternalStorage

class documentation

class ExternalStorage(WithSerializationContext): (source)

View In Hierarchy

Configuration for external storage behavior.

Warning

This API is experimental.

Method __post_init__ Validate drivers and build the internal name-keyed driver map.
Method with_context Return a copy of these options with the serialization context applied.
Class Variable driver_selector Controls which driver stores a given payload. A callable that returns the driver instance to use, or None to leave the payload stored inline. The returned driver must be one of the instances registered in ...
Class Variable drivers Drivers available for storing and retrieving payloads. At least one driver must be provided. If more than one driver is registered, driver_selector must also be set.
Class Variable payload_size_threshold Minimum payload size in bytes before external storage is considered. Defaults to 256 KiB. Set to None to consider every payload for external storage regardless of size.
Method _get_driver_by_name Looks up a driver by name, raising ValueError if not found.
Async Method _retrieve_payload Undocumented
Async Method _retrieve_payload_sequence Undocumented
Async Method _retrieve_payloads Undocumented
Method _select_driver Returns the driver to use for this payload, or None to pass through.
Async Method _store_payload Undocumented
Async Method _store_payload_sequence Undocumented
Async Method _store_payloads Undocumented
Method _validate_claim_length Undocumented
Method _validate_payload_length Undocumented
Class Variable _claim_converter Undocumented
Class Variable _context Undocumented
Class Variable _driver_map Name-keyed index of drivers, built at construction time. Used for retrieval lookups.
def __post_init__(self): (source)

Validate drivers and build the internal name-keyed driver map.

Raises ValueError if no drivers are provided, if more than one driver is registered without a driver_selector, or if any two drivers share the same name.

def with_context(self, context: SerializationContext) -> Self: (source)
overrides temporalio.converter.WithSerializationContext.with_context

Return a copy of these options with the serialization context applied.

driver_selector: Callable[[StorageDriverStoreContext, Payload], StorageDriver | None] | None = (source)

Controls which driver stores a given payload. A callable that returns the driver instance to use, or None to leave the payload stored inline. The returned driver must be one of the instances registered in drivers.

Required when more than one driver is registered. When None and only one driver is registered, that driver is used for all store operations.

drivers: Sequence[StorageDriver] = (source)

Drivers available for storing and retrieving payloads. At least one driver must be provided. If more than one driver is registered, driver_selector must also be set.

Drivers in this list are looked up by StorageDriver.name during retrieval, so each driver must have a unique name.

payload_size_threshold: int | None = (source)

Minimum payload size in bytes before external storage is considered. Defaults to 256 KiB. Set to None to consider every payload for external storage regardless of size.

def _get_driver_by_name(self, name: str) -> StorageDriver: (source)

Looks up a driver by name, raising ValueError if not found.

async def _retrieve_payload(self, payload: Payload) -> Payload: (source)

Undocumented

async def _retrieve_payload_sequence(self, payloads: Sequence[Payload]) -> list[Payload]: (source)

Undocumented

async def _retrieve_payloads(self, payloads: Payloads): (source)

Undocumented

def _select_driver(self, context: StorageDriverStoreContext, payload: Payload) -> StorageDriver | None: (source)

Returns the driver to use for this payload, or None to pass through.

async def _store_payload(self, payload: Payload) -> Payload: (source)

Undocumented

async def _store_payload_sequence(self, payloads: Sequence[Payload]) -> list[Payload]: (source)

Undocumented

async def _store_payloads(self, payloads: Payloads): (source)

Undocumented

def _validate_claim_length(self, claims: Sequence[StorageDriverClaim], expected: int, driver: StorageDriver): (source)

Undocumented

def _validate_payload_length(self, payloads: Sequence[Payload], expected: int, driver: StorageDriver): (source)

Undocumented

_claim_converter: ClassVar[JSONPlainPayloadConverter] = (source)

Undocumented

_context: SerializationContext | None = (source)

Undocumented

_driver_map: dict[str, StorageDriver] = (source)

Name-keyed index of drivers, built at construction time. Used for retrieval lookups.