Asset 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 store, which is tied to a single task instance, asset 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_store"].

When is asset_store available?

When using asset store within a task, context["asset_store"] is populated for concrete Asset inlets and outlets. A task must declare at least one concrete inlet or outlet for asset_store to contain any entries.

Accessing asset store using context

An asset becomes available through context[“asset_store”] when it is included in inlets or outlets. You can then retrieve its asset store by subscripting context[“asset_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_store = context["asset_store"][my_asset]
        watermark = asset_store.get("watermark")
        asset_store.set("watermark", "2024-06-01")

To see asset store in-action in a real DAG, checkout the DAG in example_asset_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_store"] without subscripting.

@task(inlets=[my_asset], outlets=[my_asset])
def process_single(**context):
    asset_store = context["asset_store"]
    watermark = asset_store.get("watermark")
    asset_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_store"][my_asset]) whenever a task has multiple inlets.

API reference

The following methods are available on both the per-asset accessor (context["asset_store"][my_asset]) and the shorthand (context["asset_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_store"][my_asset].get("watermark", default="initial_watermark")

set(key, value)

Writes or overwrites a key-value pair. Unlike task store, asset store has no retention parameter. Values persist until explicitly deleted or until the asset is deactivated. Like with task store, value can be any JSON-compatible type, except for None. This includes:

  • str

  • int

  • float

  • bool

  • list

  • dict

# Using context
context["asset_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_store"][my_asset].delete("watermark")

clear()

Deletes all asset store keys for the asset.

# Using context
context["asset_store"][my_asset].clear()

Some Example Use cases

Watermark pattern

The canonical use case for asset 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_store = context["asset_store"]  # single-inlet shorthand

        # Read the last watermark, default to epoch if first run.
        watermark = asset_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_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 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 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 store entries explicitly, call clear() from within a task, or use the REST API.

Note

[state_store] clear_on_success does not clear asset 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 store explicitly when it is no longer needed.

Was this entry helpful?