Asset State Store
Added in version 3.3.
Asset store is a persistent key/value store scoped to an asset, independent of any particular DAG run. Unlike task state store, which is tied to a single task instance, asset state store persists across runs and is logically owned by the asset itself. It is the natural home for cross-run metadata such as watermarks, incremental-load cursors, and per-asset configuration.
Asset store is accessed through the task context via context["asset_state_store"].
When is asset_state_store available?
When using asset state store within a task, context["asset_state_store"] is populated for concrete Asset inlets and outlets. A task must declare at least one concrete inlet or outlet for asset_state_store to contain any entries.
If a task has no inlets or outlets and accesses context["asset_state_store"], a KeyError is raised at runtime. Declare at least one asset inlet or outlet on the task to use asset state store.
This applies to both the @asset pattern and the @task pattern:
from airflow.sdk import Asset, DAG, asset, task
my_asset = Asset("my_data", uri="s3://bucket/my_data")
# @asset Dags implicitly declare the asset as an outlet
@asset
def my_asset_dag(**context):
context["asset_state_store"].set("watermark", "2024-06-01")
# @task within a @dag requires explicit inlets or outlets
with DAG("example", schedule=None):
@task(outlets=[my_asset])
def my_task(**context):
context["asset_state_store"].set("watermark", "2024-06-01")
Accessing asset state store using context
An asset becomes available through context[“asset_state_store”] when it is included in inlets or outlets. You can then retrieve its asset state store by subscripting context[“asset_state_store”] with the asset object.
from airflow.sdk import Asset, DAG, task
my_asset = Asset("my_data", uri="s3://bucket/my_data")
with DAG("example_asset_store", schedule=None):
@task(inlets=[my_asset], outlets=[my_asset])
def process(**context):
asset_state_store = context["asset_state_store"][my_asset]
watermark = asset_state_store.get("watermark")
asset_state_store.set("watermark", "2024-06-01")
To see asset state store in-action in a real DAG, checkout the DAG in example_asset_state_store.py.
Single-inlet shorthand
For tasks with exactly one concrete inlet or outlet, you can call get, set, delete, and clear directly on context["asset_state_store"] without subscripting.
@task(inlets=[my_asset], outlets=[my_asset])
def process_single(**context):
asset_state_store = context["asset_state_store"]
watermark = asset_state_store.get("watermark")
asset_state_store.set("watermark", "2024-06-01")
If the task has more than one concrete inlet or outlet, calling the shorthand raises a ValueError. Use the subscript form (context["asset_state_store"][my_asset]) whenever a task has multiple inlets.
API reference
The following methods are available on both the per-asset accessor (context["asset_state_store"][my_asset]) and the shorthand (context["asset_state_store"]) when the task has exactly one inlet.
get(key, default)
Returns the stored JSON value, or the default value if the key does not exist.
# Using context
watermark = context["asset_state_store"][my_asset].get("watermark", default="initial_watermark")
set(key, value)
Writes or overwrites a key-value pair. Unlike task state store, asset state store has no retention parameter. Values persist until explicitly deleted or until the asset is deactivated. Like with task state store, value can be any JSON-compatible type, except for None. This includes:
strintfloatboollistdict
# Using context
context["asset_state_store"][my_asset].set("watermark", "2024-06-01T00:00:00Z")
delete(key)
Deletes a single key. No-op if the key does not exist.
# Using context
context["asset_state_store"][my_asset].delete("watermark")
clear()
Deletes all asset state store keys for the asset.
# Using context
context["asset_state_store"][my_asset].clear()
Using asset_state_store inside a Watcher Trigger
BaseEventTrigger subclasses (watcher triggers) can read and write asset state store directly from within run(). The triggerer injects self.asset_state_store before run() is called, scoped to the asset the trigger is watching. It is not available during __init__ or serialize(), only access it from within run().
Unlike task-based access (where the asset is identified by an inlet or outlet declaration), the accessor in a watcher trigger is automatically bound to the watched asset, so no subscripting is needed.
import asyncio
from collections.abc import AsyncIterator
from typing import Any
from airflow.triggers.base import BaseEventTrigger, TriggerEvent
class PollEventsTrigger(BaseEventTrigger):
def __init__(self, source: str, waiter_delay: int, **kwargs):
super().__init__(**kwargs)
self.source = source
self.waiter_delay = waiter_delay
def serialize(self) -> tuple[str, dict[str, Any]]:
return (
f"{self.__class__.__module__}.{self.__class__.__qualname__}",
{"source": self.source, "waiter_delay": self.waiter_delay},
)
def _poll_for_new_record(self, source: str, last_seen: str) -> str | None:
... # Add logic for polling a certain source
return None
async def run(self) -> AsyncIterator[TriggerEvent]:
while True:
last_seen = self.asset_state_store.get("last_seen_id", default=0)
new_id = self._poll_for_new_record(
source=self.source,
last_seen=last_seen,
)
if new_id is not None:
self.asset_state_store.set("last_seen_id", new_id)
yield TriggerEvent({"status": "success", "record_id": new_id})
return
await asyncio.sleep(self.waiter_delay)
The corresponding AssetWatcher wires the trigger to the asset:
from airflow.sdk import Asset, AssetWatcher
from my_dag.triggers import PollEventsTrigger
my_asset = Asset(
name="orders_api",
watchers=[
AssetWatcher(
name="orders_api_watcher",
trigger=PollEventsTrigger(source="orders", waiter_delay=30),
)
],
)
...
self.asset_state_store behaves identically to the per-asset accessor described in the task sections above: get, set, delete, and clear are all available. Values written by the trigger are visible to any task that declares my_asset as an inlet or outlet, and vice versa.
Note
self.asset_state_store is only available inside BaseEventTrigger subclasses. Plain BaseTrigger subclasses (used for task deferral) do not have access to asset state store.
Some Example Use cases
Watermark pattern
The canonical use case for asset state store is an incremental-load task that advances a watermark on each run. The watermark is stored on the asset itself so any task that reads or writes that asset can access it. This use case is especially applicable when building things like asset “watchers” using BaseEventTrigger’s.
from airflow.sdk import Asset, DAG, task
orders = Asset("orders", uri="s3://data/orders/")
with DAG("incremental_orders", schedule="@daily"):
@task(inlets=[orders], outlets=[orders])
def load_new_orders(**context):
asset_state_store = context["asset_state_store"] # single-inlet shorthand
# Read the last watermark, default to epoch if first run.
watermark = asset_state_store.get("watermark", default="1970-01-01T00:00:00Z")
# Fetch only rows created after the watermark.
rows = fetch_orders_since(watermark)
if not rows:
return
upload_to_warehouse(rows)
# Advance the watermark to the latest row seen.
new_watermark = max(r["created_at"] for r in rows)
asset_state_store.set("watermark", new_watermark)
On each run the task reads the watermark left by the previous run, fetches only the new data, and then advances the watermark. Because asset state store persists across runs, the next run starts exactly where this one left off, even across retries, manual re-runs, or Scheduler restarts.
Lifetime and garbage collection
Asset store rows persist indefinitely. They are not subject to the [state_store] default_retention_days time-based expiry that applies to task state store.
The only automatic cleanup is an orphan sweep: when an asset is deactivated (no asset_active record exists), its store rows are removed during the next garbage-collection pass. Until that sweep runs, stale rows may exist in the database but cannot be written to. The Execution API resolver filters to active assets only.
To remove asset state store entries explicitly, call clear() from within a task, or use the REST API.
Note
[state_store] clear_on_success does not clear asset state store. Asset store is cross-run by design, so automatic task-level cleanup would destroy information that the next run depends on. Always clear asset state store explicitly when it is no longer needed.