airflow.providers.common.io.state_store.backend¶

Attributes¶

SECTION

Classes¶

StateStoreObjectStorageBackend

Object-storage backend for task and asset store.

Module Contents¶

airflow.providers.common.io.state_store.backend.SECTION = 'common.io'[source]¶

class airflow.providers.common.io.state_store.backend.StateStoreObjectStorageBackend[source]¶

Bases: airflow.sdk._shared.state.BaseStoreBackend

Object-storage backend for task and asset store.

Config keys (all under [common.io]):

state_store_objectstorage_path: base path, e.g. s3://conn_id@bucket/task-state/
state_store_objectstorage_compression: optional compression, e.g. gzip

get(scope, key, *, session=None)[source]¶

Return the stored JSON encoded value string, or None if the key does not exist.

Must handle both TaskScope and AssetScope. The execution API calls json.loads on the returned string from here, so it must be a valid JSON document.

set(scope, key, value, *, expires_at=None, session=None)[source]¶

Write or overwrite value for the given key.

Must handle both TaskScope and AssetScope. value is always a JSON encoded string (the execution API calls json.dumps before passing it here); store it verbatim so get can return it unchanged.

expires_at is an absolute UTC datetime after which the row may be deleted. Pass None (default) for a key that should never expire — stored as NULL, skipped by garbage collection.

delete(scope, key, *, session=None)[source]¶

Delete a single key. No-op if the key does not exist.

Must handle both TaskScope and AssetScope.

clear(scope, *, all_map_indices=False, session=None)[source]¶

Delete all keys under the given scope.

Must handle both TaskScope and AssetScope.

For TaskScope: by default, only keys for the exact map_index on the scope are cleared. When all_map_indices=True, the map_index filter is dropped and state is wiped across every mapped instance — for use by external callers (UI, CLI) only, not from within a running task. For AssetScope the flag has no effect.

abstract aget(scope, key, *, session=None)[source]¶

Async:

Async variant of get which returns a JSON encoded value string or None.

Must handle both TaskScope and AssetScope. session is used directly when provided; otherwise implementations manage their own session internally.

abstract aset(scope, key, value, *, expires_at=None, session=None)[source]¶

Async:

Async variant of set. value is always a JSON encoded string.

Must handle both TaskScope and AssetScope. session is used directly when provided; otherwise implementations manage their own session internally.

abstract adelete(scope, key, *, session=None)[source]¶

Async:

Async variant of delete. Must handle both TaskScope and AssetScope.

session is optional. If provided, implementations should use it directly. If None, implementations manage their own async session internally.

abstract aclear(scope, *, all_map_indices=False, session=None)[source]¶

Async:

Async variant of clear. Must handle both TaskScope and AssetScope.

For TaskScope: by default, only keys for the exact map_index on the scope are cleared. When all_map_indices=True, the map_index filter is dropped and state is wiped across every mapped instance — for use by external callers (UI, CLI) only, not from within a running task. For AssetScope the flag has no effect.

session is optional. If provided, implementations should use it directly. If None, implementations manage their own async session internally.

serialize_task_state_store_to_ref(*, value, key, scope)[source]¶

Serialize a task state store value before it is sent to the execution API for db persistence.

Called by TaskStateStoreAccessor.set() on the worker. The return value is what gets stored in the DB — typically a reference path (e.g. an S3 key) rather than the actual value. Default: return value unchanged.

Important: return only the raw reference string. The worker framework automatically wraps it in {"__airflow_state_ref__": "<ref>"} before writing to the DB, and strips that wrapper before passing stored to deserialize_task_state_store_from_ref(). Do not wrap the reference yourself.

The returned reference must be deterministic — given the same scope and key it must always return the same string. Do not use timestamps or random UUIDs as part of the reference, otherwise delete()/clear() cannot reconstruct it and the external object will be orphaned. By default, it JSON dumps the value and returns a JSON string.

deserialize_task_state_store_from_ref(stored)[source]¶

Resolve a stored task state store reference back to the actual value.

Called by TaskStateStoreAccessor.get() after the stored string is retrieved from the execution API. By default, it JSON decodes stored to reverse the default serialize_task_state_store_to_ref encoding.

serialize_asset_state_store_to_ref(*, value, key, scope)[source]¶

Serialize an asset state store value before it is sent to the Execution API for db persistence.

Called by AssetStateStoreAccessor.set() on the worker. The return value is what gets stored in the DB — typically a reference path rather than the actual value. Default: return value unchanged.

Important: return only the raw reference string. The worker framework automatically wraps it in {"__airflow_state_ref__": "<ref>"} before writing to the DB, and strips that wrapper before passing stored to deserialize_asset_state_store_from_ref(). Do not wrap the reference yourself.

The returned reference must be deterministic — given the same scope and key it must always return the same string. Do not use timestamps or random UUIDs as part of the reference, otherwise delete()/clear() cannot reconstruct it and the external object will be orphaned. By default, it JSON dumps the value and returns a JSON string.

deserialize_asset_state_store_from_ref(stored)[source]¶

Resolve a stored asset state store reference back to the actual value.

Called by AssetStateStoreAccessor.get() after the stored string is retrieved from the Execution API. By default, it JSON decodes stored to reverse the default serialize_asset_state_store_to_ref encoding.