Asset versioning

auto_code_version.py

"""Auto code_version utilities for Dagster assets.

Provides helpers that derive `code_version` strings by hashing asset function
source code at import time. This gives ZenML-style automatic cache
invalidation when code changes, without requiring manual version bumps.

CAVEATS — read before using:
  - Only hashes the decorated function body (via inspect.getsource).
  - Does NOT detect changes in imported helpers, utility modules, or
    third-party package upgrades.
  - Cosmetic changes (whitespace, comments, docstrings) WILL change the hash
    and trigger unnecessary re-materialization.
  - If the function is defined dynamically (e.g. inside a factory), the hash
    may be unstable across interpreter restarts.
  - For production ML pipelines, consider `auto_code_version_deep` which
    also hashes explicitly declared dependencies.

See dagster-io/dagster#15242 for why Dagster chose not to build this in.
"""

from __future__ import annotations

import ast
import hashlib
import inspect
import textwrap
from typing import Callable, Sequence


# ---------------------------------------------------------------------------
# Strategy 1: Shallow hash (function body only)
# ---------------------------------------------------------------------------

def auto_code_version(fn: Callable) -> str:
    """Hash the source code of *fn* and return a short hex digest.

    This is the simplest approach — equivalent to what ZenML does for step
    source code — but limited to the single function body.

    Caveats:
      * Whitespace and comment changes produce a new hash (false positive).
      * Changes to called helpers are invisible (false negative).

    Example::

        def _train_impl(features):
            ...

        @dg.asset(code_version=auto_code_version(_train_impl))
        def trained_model(features):
            return _train_impl(features)
    """
    source = inspect.getsource(fn)
    return hashlib.md5(source.encode("utf-8")).hexdigest()[:12]


# ---------------------------------------------------------------------------
# Strategy 2: Normalized hash (strips comments & whitespace)
# ---------------------------------------------------------------------------

def auto_code_version_normalized(fn: Callable) -> str:
    """Hash a normalized AST of *fn*, ignoring whitespace and comments.

    Parses the function source into an AST, strips docstrings, then dumps
    the tree back to a canonical string. This avoids false positives from
    formatting-only changes while still catching logic changes.

    Caveats:
      * Still only covers the single function body.
      * Changes to called helpers are invisible (false negative).
      * Renaming local variables WILL change the hash (correct behavior).

    Example::

        @dg.asset(code_version=auto_code_version_normalized(my_func))
        def my_asset(): ...
    """
    source = textwrap.dedent(inspect.getsource(fn))
    tree = ast.parse(source)

    # Strip docstrings from function defs
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if (
                node.body
                and isinstance(node.body[0], ast.Expr)
                and isinstance(node.body[0].value, (ast.Constant, ast.Str))
            ):
                node.body.pop(0)

    canonical = ast.dump(tree)
    return hashlib.md5(canonical.encode("utf-8")).hexdigest()[:12]


# ---------------------------------------------------------------------------
# Strategy 3: Deep hash (function + explicit deps)
# ---------------------------------------------------------------------------

def auto_code_version_deep(
    fn: Callable,
    deps: Sequence[Callable] = (),
    extra: str = "",
) -> str:
    """Hash *fn* plus every callable in *deps* and an optional *extra* salt.

    This is the closest analog to ZenML's full cache key, which hashes step
    source + input artifact IDs + parameters. Here you explicitly list the
    helper functions and can add a salt for package versions, config, etc.

    Args:
        fn: The primary asset function.
        deps: Additional callables whose source should be included in the
            hash (imported helpers, transformers, model classes, etc.).
        extra: Arbitrary string folded into the hash — use for package
            versions, config hashes, or feature-flag state.

    Caveats:
      * You must manually list deps — nothing is auto-discovered.
      * C-extension functions (e.g. numpy ufuncs) have no inspectable source
        and will raise TypeError; pass their __version__ in *extra* instead.

    Example::

        from mylib import preprocess, build_features
        import sklearn

        @dg.asset(
            code_version=auto_code_version_deep(
                _train_impl,
                deps=[preprocess, build_features],
                extra=sklearn.__version__,
            )
        )
        def trained_model(features):
            return _train_impl(features)
    """
    h = hashlib.md5()
    h.update(inspect.getsource(fn).encode("utf-8"))

    for dep in deps:
        try:
            h.update(inspect.getsource(dep).encode("utf-8"))
        except (TypeError, OSError):
            # Built-in or C-extension — include repr as best-effort fallback
            h.update(repr(dep).encode("utf-8"))

    if extra:
        h.update(extra.encode("utf-8"))

    return h.hexdigest()[:12]


# ---------------------------------------------------------------------------
# Strategy 4: ZenML-like decorator — zero-config auto-versioned eager asset
# ---------------------------------------------------------------------------

def cached_asset(
    fn: Callable | None = None,
    *,
    deps: Sequence[Callable] = (),
    extra: str = "",
    normalize: bool = True,
    **asset_kwargs,
):
    """Decorator that wraps ``@dg.asset`` with ZenML-like behavior.

    Automatically sets:
      - ``code_version`` from source hash (normalized AST by default)
      - ``automation_condition=AutomationCondition.eager()`` for transitive
        propagation of upstream changes

    Together these two settings make Dagster behave like ZenML's cached steps:
      - Code changes → asset marked stale → auto-rematerialized
      - Upstream re-materialized → downstream marked stale → auto-rematerialized
      - No changes → asset is not touched

    Args:
        fn: The asset function (when used without parentheses).
        deps: Helper callables to include in the code hash.
        extra: Extra salt (package versions, config) for the hash.
        normalize: If True (default), use AST-normalized hashing to ignore
            whitespace/comment changes. If False, use raw source hash.
        **asset_kwargs: Passed through to ``@dg.asset`` (group_name, owners,
            tags, key_prefix, etc.).  ``code_version`` and
            ``automation_condition`` are set automatically but can be
            overridden explicitly.

    Caveats:
      * Same inspect.getsource limitations as other strategies.
      * ``eager()`` requires the ``default_automation_condition_sensor`` to
        be enabled in the Dagster UI under Automation → Sensors.
      * Transitive propagation depends on ``eager()`` firing on each layer
        in sequence — there may be a ~30 s evaluation delay per hop.

    Example — bare decorator::

        @cached_asset
        def features(raw_data):
            ...

    Example — with options::

        @cached_asset(deps=[preprocess], extra=sklearn.__version__,
                      group_name="ml")
        def trained_model(features):
            ...
    """
    import dagster as dg

    def _wrap(func: Callable):
        # Compute code_version if not explicitly overridden
        if "code_version" not in asset_kwargs:
            if deps or extra:
                version = auto_code_version_deep(func, deps=deps, extra=extra)
            elif normalize:
                version = auto_code_version_normalized(func)
            else:
                version = auto_code_version(func)
            asset_kwargs["code_version"] = version

        # Set eager() if not explicitly overridden
        if "automation_condition" not in asset_kwargs:
            asset_kwargs["automation_condition"] = (
                dg.AutomationCondition.eager()
            )

        return dg.asset(**asset_kwargs)(func)

    # Support both @cached_asset and @cached_asset(...)
    if fn is not None:
        return _wrap(fn)
    return _wrap

Versioning_assets.md

Dagster Asset Versioning & Staleness for ML Pipelines

1. code_version: Manually Set, Not Auto-Hashed

code_version is a manually assigned string on the @asset decorator or AssetSpec. Dagster does not automatically hash your function source code.

import dagster as dg

@dg.asset(code_version="1.0.0")
def trained_model(features):
    ...

There is an open feature request (dagster-io/dagster#15242) for automatic source-code hashing (e.g. CodeVersion.auto()), but it was not implemented. Dagster maintainers intentionally chose not to build this because:

• Simple inspect.getsource() hashing is fragile — it misses imported functions, package version changes, and external dependencies.
• False negatives: a real code change in an imported utility wouldn't bump the version.
• False positives: a cosmetic change (whitespace, comment) would invalidate all downstream assets.

DIY Auto-Hashing — Three Strategies (see auto_code_version.py)

We provide a utility module auto_code_version.py with three strategies of increasing coverage. Each trades off convenience against false-positive/false-negative risk.

Strategy 1: Shallow hash (auto_code_version)

Hashes raw function source via inspect.getsource() — the simplest approach.

from auto_code_version import auto_code_version

def _train_impl(features):
    model = fit(features)
    return model

@dg.asset(code_version=auto_code_version(_train_impl))
def trained_model(features):
    return _train_impl(features)

Triggers a change	Does NOT trigger a change
Any edit to the function body	Changes to imported helpers (from mylib import preprocess)
Whitespace / comment changes (false positive)	Package version upgrades (sklearn 1.4→1.5)
Renamed variables	Environment variable / config changes
	External schema changes

Strategy 2: Normalized hash (auto_code_version_normalized)

Parses the function into an AST, strips docstrings and whitespace, then hashes the canonical tree. Avoids false positives from formatting-only changes.

from auto_code_version import auto_code_version_normalized

@dg.asset(code_version=auto_code_version_normalized(my_func))
def my_asset(): ...

Triggers a change	Does NOT trigger a change
Logic changes, new/removed statements	Whitespace, comment, docstring edits
Renamed variables	Changes to imported helpers
Changed constants/literals	Package upgrades

Strategy 3: Deep hash (auto_code_version_deep)

Hashes the function plus explicitly listed dependency callables and an arbitrary salt string. This is the closest analog to ZenML's full cache key.

from auto_code_version import auto_code_version_deep
from mylib import preprocess, build_features
import sklearn

@dg.asset(
    code_version=auto_code_version_deep(
        _train_impl,
        deps=[preprocess, build_features],
        extra=sklearn.__version__,
    )
)
def trained_model(features):
    return _train_impl(features)

Triggers a change	Does NOT trigger a change
Edit to main function body	Unlisted helper changes (false negative)
Edit to any listed deps function	C-extension functions (falls back to repr())
Change in extra string (e.g. package version)	Env vars, config, external schemas

Caveat Summary For All Strategies

• Evaluated at import time — the hash is computed when the module loads, not at materialization time.
• inspect.getsource() requires source files — won't work on functions defined in a REPL, compiled .pyc-only distributions, or C extensions.
• Dynamic functions (e.g. generated inside a factory loop) may produce unstable hashes across interpreter restarts.
• For production, Dagster maintainers recommend manual code_version strings (semver or date-based) as the most reliable approach — see dagster-io/dagster#15242 for the full rationale.

---

2. How Dagster Computes Versions Internally

Dagster has two version concepts:

Concept	Who Sets It	What It Represents
code_version	You (manually)	Version of the computation logic
data_version	Dagster (auto-generated) or you (via DataVersion)	Fingerprint of the asset's output value

Data Version (Logical Version)

Dagster auto-generates a data version (also called "logical version") by hashing together:

1. The asset's code_version
2. The data_version of each input (upstream asset)

This means: if neither your code nor your inputs changed, the data version stays the same, and Dagster knows the output would be identical — enabling caching/skip behavior.

---

3. The "Unsynced" Label — Manual Staleness Detection in the UI

What Causes an Asset to Show "Unsynced"

An asset is marked Unsynced in the Dagster UI when any of these are true:

1. Code version changed — the code_version on the asset definition differs from the code_version recorded during its last materialization.
2. Dependencies changed — an upstream dependency was added or removed since the last materialization.
3. Upstream data changed — a direct parent asset was re-materialized (producing a new data version) after the downstream asset's last materialization.

Non-Transitive Propagation (v1.8.0+)

As of Dagster 1.8.0, Unsynced is NOT transitive. This is a deliberate design choice:

A (code_version changed → Unsynced)
└── B (direct child → Unsynced, because A was re-materialized after B)
    └── C (grandchild → NOT Unsynced, until B is actually re-materialized)

• Only direct children of a changed/re-materialized asset show as Unsynced.
• Grandchildren and deeper descendants do not cascade to Unsynced automatically.
• This prevents "Unsynced label fatigue" across large graphs and improves UI performance.

Using "Materialize Unsynced" in the UI

The Dagster UI provides a "Materialize Unsynced" button that selects all assets currently labeled Unsynced and launches a materialization run for them. This is the manual workflow for selective re-computation:

1. Change code_version on Step 4 (e.g. "1.0.0" → "1.1.0")
2. Redeploy your code location
3. Open the asset graph in the Dagster UI
4. Step 4 shows the Unsynced label; Steps 1–3 do not
5. Click "Materialize Unsynced" — only Step 4 runs

---

4. Observable Source Assets and DataVersion

For external data sources (files, APIs, databases you don't control), use observable source assets to report data versions:

import dagster as dg
import hashlib

@dg.observable_source_asset
def raw_training_data():
    """Observe an external CSV file and report its data version."""
    content = open("/data/training.csv", "rb").read()
    version = hashlib.sha256(content).hexdigest()[:16]
    return dg.DataVersion(version)

• Click "Observe Sources" in the UI (or run via a sensor/schedule) to check for new data versions.
• If the returned DataVersion differs from the last observation, downstream assets are marked Unsynced.
• This is how Dagster detects upstream data changes for external sources without re-materializing them.

---

5. Using Versioning with Jobs and Schedules (stale_assets_only)

Declarative Automation isn't the only option. You can use traditional jobs + schedules with the stale_assets_only parameter on RunRequest:

import dagster as dg

ml_pipeline_job = dg.define_asset_job(
    "ml_pipeline_job",
    selection=[raw_data, features, trained_model, evaluation],
)

@dg.schedule(
    cron_schedule="0 2 * * *",  # Daily at 2 AM
    job=ml_pipeline_job,
)
def nightly_ml_refresh():
    """Only re-materialize assets that are actually stale."""
    return dg.RunRequest(stale_assets_only=True)

How stale_assets_only=True Works

• The schedule fires on the cron tick as usual.
• Dagster evaluates which assets in the job's selection are stale (Unsynced).
• Only stale assets are included in the run. Non-stale assets are skipped entirely.
• If no assets are stale, no run is launched.
• If passed without an asset selection, all stale assets in the job are materialized.

Combining With Observable Source Assets

For a full ML pipeline with external data detection:

@dg.observable_source_asset
def training_data_source():
    version = compute_hash_of_external_data()
    return dg.DataVersion(version)

@dg.asset(code_version="1.0.0")
def features(training_data_source): ...

@dg.asset(code_version="1.0.0")
def trained_model(features): ...

@dg.asset(code_version="1.0.0")
def evaluation(trained_model): ...

# Observe sources on a schedule, then materialize stale downstream
observe_job = dg.define_asset_job(
    "observe_sources", selection=dg.AssetSelection.all_asset_checks()
)

ml_job = dg.define_asset_job(
    "ml_pipeline", selection=[features, trained_model, evaluation]
)

@dg.schedule(cron_schedule="0 1 * * *", job=observe_job)
def observe_schedule():
    return dg.RunRequest()

@dg.schedule(cron_schedule="0 2 * * *", job=ml_job)
def ml_schedule():
    return dg.RunRequest(stale_assets_only=True)

---

Scenario: "Only Step 4 code changed" — Both Systems

ZenML:

Run pipeline →
  Step 1: cache HIT (same code + same inputs) → skip, reuse output
  Step 2: cache HIT → skip, reuse output
  Step 3: cache HIT → skip, reuse output
  Step 4: cache MISS (source code hash changed) → execute

All four steps are attempted but three are skipped via cache lookup. The pipeline run still appears with all four steps, three marked "cached."

Dagster (with stale_assets_only):

Schedule fires →
  Dagster checks staleness: only Step 4 is Unsynced
  Run is launched with ONLY Step 4 in the selection
  Steps 1–3 are not part of the run at all

Steps 1–3 never enter a run. The run contains only Step 4.

---

6. Making Dagster Behave Like ZenML: @cached_asset

The gap between Dagster and ZenML boils down to two missing defaults:

1. No auto code hashing — you must set code_version manually
2. No auto propagation — the Unsynced label doesn't cascade transitively

We can close both gaps with the @cached_asset decorator (see auto_code_version.py), which combines auto-hashed code_version with AutomationCondition.eager():

Before (manual Dagster)

import dagster as dg

@dg.asset(code_version="1.0.0")  # must remember to bump
def features(raw_data): ...

@dg.asset(code_version="1.0.0")
def trained_model(features): ...

@dg.asset(code_version="1.1.0")  # forgot to bump? silent staleness
def evaluation(trained_model): ...

After (@cached_asset — ZenML-like)

from auto_code_version import cached_asset

@cached_asset
def features(raw_data):
    return engineer_features(raw_data)

@cached_asset
def trained_model(features):
    return fit_model(features)

@cached_asset
def evaluation(trained_model):
    return evaluate(trained_model)

That's it. No manual version strings. Under the hood, each asset gets:

• code_version = AST-normalized hash of the function source
• automation_condition = AutomationCondition.eager()

How This Achieves ZenML-Like Behavior

ZenML behavior	How @cached_asset replicates it
Code change → cache miss	Source hash changes → code_version changes → asset marked Unsynced
Upstream reruns → downstream cache miss	eager() fires when any upstream re-materializes → transitive propagation
No change → skip	Code hash unchanged + no upstream update → asset stays synced, not executed
Zero config	Just @cached_asset — no version strings to manage

The Transitive Propagation Solution

The Unsynced UI label is non-transitive (stops at direct children). But eager() Declarative Automation IS effectively transitive because it works in a chain:

A changes → A re-materializes
         → B sees upstream update → eager() fires → B re-materializes
                                 → C sees upstream update → eager() fires → C re-materializes

Each hop takes ~30 seconds (the automation sensor evaluation interval), so a 4-step pipeline cascades in ~90 seconds. This is slower than ZenML (which resolves within a single run), but the result is the same: only changed assets and their true downstream dependents re-execute.

Advanced: Adding Helper Deps and Package Versions

from auto_code_version import cached_asset
from mylib import preprocess, build_features
import sklearn

@cached_asset(deps=[preprocess, build_features], extra=sklearn.__version__)
def trained_model(features):
    preprocessed = preprocess(features)
    X = build_features(preprocessed)
    return sklearn.ensemble.RandomForestClassifier().fit(X, y)

Now changes to preprocess, build_features, or an sklearn upgrade all trigger re-materialization.

Prerequisites

• Enable the automation sensor: Toggle on default_automation_condition_sensor in the Dagster UI under Automation → Sensors. Without this, eager() conditions are never evaluated.
• Assets must be in the same code location for same-run grouping. Cross-location assets cascade via separate runs.

Remaining Gaps vs ZenML

Even with @cached_asset, Dagster still differs from ZenML in these ways:

Gap	Why it exists	Workaround
Parameters not in hash	Dagster code_version is static at import time; ZenML hashes runtime params	Pass config values in the extra salt
~30s delay per hop	eager() evaluates on a sensor tick interval	Acceptable for ML pipelines (training takes minutes/hours anyway)
inspect.getsource limitations	Can't hash C extensions, dynamic code	Use extra for package __version__ strings
No artifact-store-aware isolation	ZenML scopes cache to workspace + artifact store	Dagster's Unsynced is scoped to the instance; use separate deployments for isolation

---

Sources

Dagster

• Asset Versioning and Caching — Dagster Docs
• Unsynced Status Propagation Discussion (dagster-io/dagster#25248)
• Auto Code Versioning Feature Request (dagster-io/dagster#15242)
• Materialize Stale Assets on Schedule (dagster-io/dagster#14755)
• Support Rematerialization of Stale Assets in Schedules (dagster-io/dagster#10726)
• Partitioned Assets and code_version (dagster-io/dagster#22704)
• Declarative Scheduling Blog Post

ZenML

• ZenML Caching — Control Caching Behavior
• ZenML Advanced Step/Pipeline Features
• ZenML Artifacts Concepts
• Why You Should Be Using Caching in ML Pipelines — ZenML Blog
• ZenML Steps SDK Reference (cache key internals)