Dagster Expert

Assets can be selected two ways:

• String-based selection syntax — works identically in the UI search bar, dg CLI (--assets), and dg.AssetSelection.from_coercible() in Python
• `AssetSelection` in Python — programmatic API with set operations (|, &, -), traversals, and methods not available in string syntax

String-Based Selection Syntax

Attributes

• key:<name> or just <name> — select by asset key (e.g. customers)
• tag:<key>=<value> or tag:<key> — select by tag (e.g. tag:priority=high)
• owner:<value> — select by owner (e.g. owner:team@company.com)
• group:<value> — select by group (e.g. group:sales_analytics)
• kind:<value> — select by kind (e.g. kind:dbt)
• code_location:<value> — select by code location (e.g. code_location:my_project)
• status:<value> — select by materialization status
• column:<value> — select by column name (assets with table schema metadata)
• table_name:<value> — select by table name
• column_tag:<key>=<value> or column_tag:<key> — select by column-level tag
• changed_in_branch:<value> — select assets changed in a git branch (Dagster Plus)

Wildcards: key:customer*, key:*_raw, * (all assets)

Operators

• and / AND — e.g. tag:priority=high and kind:dbt
• or / OR — e.g. group:sales or group:marketing
• not / NOT — e.g. not kind:dbt
• (expr) — grouping, e.g. tag:priority=high and (kind:dbt or kind:python)

Functions

• sinks(expr) — assets with no downstream dependents (e.g. sinks(group:analytics))
• roots(expr) — assets with no upstream dependencies (e.g. roots(kind:dbt))

Traversals

• +expr — all upstream dependencies (e.g. +customers)
• expr+ — all downstream dependents (e.g. customers+)
• N+expr — N levels upstream (e.g. 2+kind:dbt)
• expr+N — N levels downstream (e.g. group:sales+1)
• N+expr+M — N up, M down (e.g. 1+key:customers+2)

Examples

Selection strings (work identically in UI, CLI, and Python):

# By metadata
tag:priority=high and kind:dbt
group:sales or group:marketing
not kind:dbt
owner:team@company.com

# With traversals
+kind:dbt                        # all upstream of dbt assets
group:sales+                     # group:sales + all downstream
2+key:customers                  # customers + 2 levels upstream

# With functions
sinks(group:analytics)           # terminal assets in group
roots(kind:dbt)                  # source dbt assets

Using in the CLI:

dg launch --assets "tag:priority=high and kind:dbt"
dg list defs --assets "group:sales"

Using in Python (via from_coercible):

sel = dg.AssetSelection.from_coercible("tag:priority=high and kind:dbt")

---

Python API

Parsing Selection Strings

dg.AssetSelection.from_coercible() converts a selection string (or other coercible types) into an AssetSelection object. It accepts:

• A selection string (parsed using the same grammar as the UI and CLI)
• An existing AssetSelection instance (returned as-is)
• A sequence of strings (each parsed and unioned together)
• A sequence of AssetsDefinition or AssetKey objects

# Parse a selection string
sel = dg.AssetSelection.from_coercible("tag:priority=high and kind:dbt")

# Pass to APIs that expect AssetSelection
job = dg.define_asset_job("my_job", selection=sel)

Basic Selection

# Select specific assets
dg.AssetSelection.assets("asset_a", "asset_b", "asset_c")

# Select all assets
dg.AssetSelection.all()

# Select by group
dg.AssetSelection.groups("analytics", "raw_data")

# Select by tag
dg.AssetSelection.tag("priority", "high")

Dependency-Based Selection

# Select asset and all upstream dependencies
dg.AssetSelection.assets("final_report").upstream()

# Select asset and all downstream dependencies
dg.AssetSelection.assets("raw_data").downstream()

# Select asset and immediate upstream only
dg.AssetSelection.assets("final_report").upstream(depth=1)

Combining Selections

selection_a = dg.AssetSelection.assets("a")
selection_b = dg.AssetSelection.assets("b")

# Union: assets in A OR B
selection_a | selection_b

# Intersection: assets in A AND B
selection_a & selection_b

# Difference: assets in A but not in B
selection_a - selection_b

# Example: All analytics assets except one
dg.AssetSelection.groups("analytics") - dg.AssetSelection.assets("excluded_asset")

Using in Jobs

analytics_job = dg.define_asset_job(
    name="analytics_job",
    selection=dg.AssetSelection.groups("analytics").downstream(),
)

---

Python-Only Methods

These methods are only available via the Python API and have no string syntax equivalent:

• dg.AssetSelection.key_prefixes(["warehouse", "staging"]) — select by key prefix (key:prefix* in string syntax is a partial alternative)
• dg.AssetSelection.key_substring("customer") — select by substring match on asset key
• selection.required_multi_asset_neighbors() — include co-selected assets in non-subsettable multi-asset definitions
• selection.materializable() — filter to only materializable (non-observable, non-external) assets
• selection.upstream_source_assets() — select external/source assets that are upstream parents
• selection.without_checks() — remove asset checks from a selection
• dg.AssetSelection.checks_for_assets("my_asset") — select asset checks targeting specific assets
• dg.AssetSelection.checks(my_check_key) — select specific asset checks by key
• dg.AssetSelection.all_asset_checks() — select all asset checks

Advanced Asset Patterns

@multi_asset

Use when a single computation produces multiple assets. Define outputs with specs=[...] using AssetSpec, and yield MaterializeResult for each asset.

```python nocheckundefined @dg.multi_asset( specs=[dg.AssetSpec("users"), dg.AssetSpec("orders", deps=["users"])], ) def load_data(): users_df = fetch_users() yield dg.MaterializeResult(asset_key="users", metadata={"row_count": len(users_df)})

orders depend on user data for enrichment

orders_df = fetch_orders(users_df) yield dg.MaterializeResult(asset_key="orders", metadata={"row_count": len(orders_df)})

When to use:

- One computation produces multiple related assets
- Assets share expensive setup or have computational dependencies on each other

**Subsettability:** By default, all assets in a `@multi_asset` are materialized together. To allow materializing a subset, set `can_subset=True` on the decorator and `skippable=True` on individual `AssetSpec`s. Use `context.op_execution_context.selected_asset_keys` to check which assets were requested.

**Static metadata on specs:** `AssetSpec` accepts the same metadata parameters as `@dg.asset` — `description`, `group_name`, `owners`, `tags`, `kinds`, `deps`, `code_version`, `automation_condition`, etc. See [Asset Definition Properties](./definition-metadata.md) for details on each parameter.

## MaterializeResult

`MaterializeResult` records dynamic metadata each time an asset materializes. Use it as a return type for `@dg.asset` or yield it in `@multi_asset`.

@dg.asset def my_asset() -> dg.MaterializeResult: data = [...] return dg.MaterializeResult( metadata={ "row_count": dg.MetadataValue.int(len(data)), "last_updated": dg.MetadataValue.text(str(datetime.now())), "sample_data": dg.MetadataValue.json(data[:5]), } )

`MaterializeResult[T]` can also carry a value (like `Output[T]`), making it the preferred return type for greenfield code:

@dg.asset def my_asset() -> dg.MaterializeResult[dict]: data = {"key": "value"} return dg.MaterializeResult(value=data, metadata={"size": len(data)})

### MetadataValue Types

- `MetadataValue.int(n)` — integer values (row counts)
- `MetadataValue.float(n)` — float values (percentages)
- `MetadataValue.text(s)` — short text values
- `MetadataValue.json(obj)` — JSON-serializable objects
- `MetadataValue.md(s)` — markdown text
- `MetadataValue.url(s)` — clickable URLs
- `MetadataValue.path(s)` — file paths
- `MetadataValue.table(records)` — tabular data

## @graph_asset

Compose multiple `@op`s into a single asset. Each op is independently retriable — if the last step fails, you can retry without re-running earlier steps.

@dg.op def fetch_data() -> dict: return {"raw": [1, 2, 3]}

@dg.op def transform_data(data: dict) -> dict: return {"processed": [x * 2 for x in data["raw"]]}

@dg.graph_asset def complex_asset(): raw = fetch_data() return transform_data(raw)

When to use:

- Single asset requires multiple distinct steps
- You want independent retriability for each step
- Steps are reusable across multiple assets

## @graph_multi_asset

Combine `@graph_asset` and `@multi_asset` — compose ops into a pipeline that produces multiple assets.

@dg.graph_multi_asset( outs={ "users": dg.AssetOut(), "orders": dg.AssetOut(), } ) def etl_pipeline(): raw_data = extract_from_api() cleaned = clean_data(raw_data) return {"users": extract_users(cleaned), "orders": extract_orders(cleaned)}

When to use:

- Multiple assets require shared complex multi-step logic
- Steps are expensive and should be shared
- Better encapsulation than separate assets with `deps=`

Asset Definition Properties

Decorator Parameters

Applied once when the asset is defined:

@dg.asset(
    description="Detailed description for the UI",
    group_name="analytics",
    key_prefix=["warehouse", "staging"],
    owners=["team:data-engineering", "user@example.com"],
    tags={"priority": "high", "pii": "true", "domain": "sales"},
    code_version="1.2.0",
)
def my_asset() -> None:
    pass

• owners — specify team (team:name) or individuals for accountability
• tags — primary organizational mechanism; use liberally for filtering and grouping (also used by asset selection and automation conditions)
• code_version — track when asset logic changes for lineage and debugging
• description — explain what the asset represents and its business purpose (docstring also works)
• group_name — visual organization in UI; use for data layers or domains
• key_prefix — generates the asset key as AssetKey([*prefix, fn_name]), e.g. key_prefix=["warehouse", "raw"] on a function named orders produces AssetKey(["warehouse", "raw", "orders"]). Use the name argument to override the function name portion (useful in factory patterns that produce many assets from one function).

Setting Properties on AssetSpec

For @multi_asset, set the same properties on each AssetSpec:

```python nocheckundefined @dg.multi_asset( specs=[ dg.AssetSpec( "users", group_name="raw_data", owners=["team:data-engineering"], tags={"priority": "high"}, description="Raw user records from API", ), dg.AssetSpec( "orders", group_name="raw_data", deps=["users"], ), ], ) def load_data(): ...

`AssetSpec` accepts the same metadata parameters as `@dg.asset`: `description`, `group_name`, `owners`, `tags`, `kinds`, `deps`, `code_version`, `automation_condition`, `key_prefix`, and more.

Asset Dependencies

Parameter-Based Dependencies

When an asset depends on another Dagster-managed asset, add it as a function parameter. Dagster uses an IOManager to load the upstream asset's output into memory and pass it as a Python object.

@dg.asset
def upstream_asset() -> dict:
    return {"data": [1, 2, 3]}

@dg.asset
def downstream_asset(upstream_asset: dict) -> list:
    # upstream_asset is loaded into memory via IOManager
    return upstream_asset["data"]

• Parameter name must match the upstream asset's function name (or asset key)
• Dagster automatically materializes upstream first, then loads and passes the output
• Creates a visible dependency edge in the asset graph
• Use when you want Dagster to manage data transfer between assets

deps= Dependencies

Use deps= to declare a data dependency for lineage and scheduling purposes only. The asset function does NOT receive the upstream data. Either the function itself handles data access (e.g. reading from a database directly), or some external process ensures the data is available.

@dg.asset(deps=["external_table", "raw_file"])
def processed_data() -> None:
    # No upstream values passed in — read from sources directly
    pass

• Declares ordering and lineage without coupling data transfer
• Use when the upstream asset doesn't return a value, is external, or data is managed outside Dagster's IOManager system
• The dependency still affects scheduling: Dagster knows processed_data should run after external_table

Mixed Dependencies

Combine both patterns when an asset has some IOManager-managed inputs and some loose data dependencies:

@dg.asset(deps=["raw_file"])
def enriched_data(reference_table: dict) -> dict:
    # reference_table: loaded via IOManager (parameter-based)
    # raw_file: declared dependency only, read manually
    return {"enriched": reference_table}

Asset Patterns

When to Use Each Pattern

• Basic `@dg.asset` — simple one-to-one transformation
• Parameter-based dependency — asset depends on another managed asset, data loaded via IOManager
• `deps=` dependency — asset depends on external or non-Python asset, data dependency only
• `@multi_asset` — single operation produces multiple related assets
• `@graph_asset` — multiple op steps needed to produce one asset
• `@graph_multi_asset` — complex pipeline producing multiple assets from composed ops
• Asset factory — generate many similar assets programmatically

Quick Notes on Basic Patterns

Basic `@dg.asset`: Function name becomes the asset key. Docstring becomes the description in the UI. Return type annotation is optional but recommended.

Asset groups: Use group_name= on the decorator to organize assets visually in the UI. Common groupings: by data layer (raw, staging, analytics), by domain (sales, marketing), or by source (postgres, api).

Key prefixes: Use key_prefix=["warehouse", "raw"] to namespace asset keys hierarchically (e.g. warehouse/raw/orders). Useful for multi-tenant or layered architectures.

Configuration: Subclass dg.Config with typed fields, then add as a parameter to your asset function. Fields become configurable at launch time.

Execution context: Add context: dg.AssetExecutionContext as a parameter to access context.log, context.asset_key, context.partition_key (if partitioned), and context.run_id.

Return types: Assets can return data directly (passed to downstream via IOManager) or dg.MaterializeResult (for metadata, or dg.MaterializeResult[T] for data + metadata). MaterializeResult[T] is preferred over dg.Output[T] in greenfield code.

Common Anti-Patterns

• Verb-based names like load_customers — use nouns describing the output: customers
• Giant asset doing everything — split into focused, composable assets
• No type annotations — add a return type: -> dict, -> None
• No docstring — add a description via docstring or description=
• Ignoring `MaterializeResult` — return metadata for observability

Reference Files

<!-- BEGIN GENERATED INDEX -->

• Advanced Asset Patterns — @multi_asset, @graph_asset, @graph_multi_asset, or asset factories; MaterializeResult, dynamic metadata, MetadataValue types
• Asset Definition Properties — asset metadata, tags, owners, groups, key_prefix, code_version, AssetSpec properties
• Asset Dependencies — asset dependencies, parameter-based deps, deps= external dependencies

<!-- END GENERATED INDEX -->

Choosing an Automation Approach

Dagster provides three main approaches to automation: schedules for time-based execution, sensors for event-driven triggers, and declarative automation for asset-centric condition-based orchestration.

Workflow Decision Tree

Choose your automation approach based on your use case:

• Simple, fixed time-based execution → Schedules
• Custom polling logic → Basic Sensors
• Launching jobs in response to asset materialization events → Asset Sensors
• Triggering compute based on run success/failure → Run Status Sensors
• Partition-aware scheduling, declarative/asset-based scheduling, scheduling depending on asset graph state and materialization events → Declarative Automation

Core Concepts

Jobs

A job is a selection of assets to execute together. Jobs are the unit of execution that schedules and sensors trigger.

import dagster as dg

# Define a job that selects specific assets
analytics_job = dg.define_asset_job(
    name="analytics_job",
    selection=["sales_data", "customer_metrics"]
)

Jobs can also select assets by tags, groups, or patterns:

# Select all assets with a specific tag
tagged_job = dg.define_asset_job(
    name="daily_job",
    selection=dg.AssetSelection.tag("priority", "high")
)

# Select all assets in a group
group_job = dg.define_asset_job(
    name="etl_job",
    selection=dg.AssetSelection.groups("etl")
)

Automation Approaches

Schedules: Time-based execution with cron expressions. Best for predictable, recurring tasks.

Sensors: Poll for external events and trigger runs. Best for file arrivals, API events, or custom conditions.

Declarative Automation: Set conditions directly on assets. Best for complex dependency logic and asset-centric workflows. Automatic handling of asset and partition state and dependencies.

Declarative Automation: Advanced Concepts

This document covers advanced topics for deep understanding of the declarative automation system.

Status vs Events

Understanding the distinction between statuses and events is fundamental to building correct automation conditions.

Statuses

Statuses are persistent conditions that remain true for multiple evaluation ticks.

Examples:

• AutomationCondition.missing() - Stays true until the partition is materialized
• AutomationCondition.in_progress() - True while a run is executing
• AutomationCondition.in_latest_time_window() - True for the latest time partition(s)

Characteristic: If the underlying state doesn't change, the status will be true for consecutive evaluations.

Events

Events are transient conditions that are true only on a single evaluation tick.

Examples:

• AutomationCondition.newly_updated() - True only on the tick when materialization occurs
• AutomationCondition.code_version_changed() - True only on the first tick after code changes
• AutomationCondition.cron_tick_passed() - True only on the first tick after the cron tick

Characteristic: Even if evaluated immediately again, the event would be false (assuming no new change).

Converting Between Status and Event

Status → Event with `newly_true()`:

# missing() is a status (stays true for many ticks)
# newly_true() converts it to an event (true only when becoming missing)
condition = dg.AutomationCondition.missing().newly_true()

Use case: Prevent repeated requests during persistent states. A partition stays missing while a run is in progress. Using newly_true() ensures you only request it once.

Two Events → Status with `since()`:

# Both newly_updated() and newly_requested() are events
# since() converts them to a status: "updated more recently than requested"
condition = dg.AutomationCondition.newly_updated().since(
    dg.AutomationCondition.newly_requested()
)

Use case: Create persistent states from transient events. This condition becomes true when an update occurs and stays true until a request is made.

Example: Preventing Duplicate Requests

The default eager() condition uses this pattern:

(
    dg.AutomationCondition.newly_missing()
    | dg.AutomationCondition.any_deps_updated()
).since_last_handled()

• newly_missing() and any_deps_updated() are events
• since_last_handled() converts them to a status that persists until the asset is requested or updated
• Without this conversion, the condition would only be true for a single tick, potentially missing the opportunity to launch a run

Run Grouping

Run grouping allows multiple assets to execute in a single run even though downstream assets' dependencies haven't been materialized yet.

The Problem

Consider assets A → B → C, all with eager() conditions:

1. A's upstream updates, triggering A 2. A is requested and begins executing 3. On the next tick, B sees that A hasn't finished materializing 4. Without run grouping, B would wait for A to complete 5. This results in three separate runs instead of one

The Solution: will_be_requested()

The will_be_requested() operand is true for assets that will be requested in the current tick. Dependency conditions use this to group assets:

# From any_deps_updated() definition:
dg.AutomationCondition.any_deps_match(
    (
        dg.AutomationCondition.newly_updated()
        & ~dg.AutomationCondition.executed_with_root_target()
    )
    | dg.AutomationCondition.will_be_requested()  # Enables run grouping
)

When evaluating B:

1. B checks if any dependencies are updated OR will be requested this tick 2. A is marked as "will be requested" this tick 3. B treats A as if it were already updated 4. B is also marked for execution in the same run as A

Requirements for Same-Run Execution

Two assets can execute in the same run if:

1. Same repository: They must be in the same code location 2. Compatible partitions: They must have matching PartitionsDefinition objects 3. Compatible partition mapping: Must use TimeWindowPartitionMapping or IdentityPartitionMapping

If these requirements aren't met, assets execute in separate runs even with run grouping logic.

Dependency Filtering with allow() and ignore()

Dependency operators (any_deps_match(), all_deps_match()) check conditions on upstream assets. Filtering controls which upstreams are checked.

allow() Creates Intersection

Only dependencies in the selection are checked:

condition = dg.AutomationCondition.any_deps_match(
    dg.AutomationCondition.missing()
).allow(dg.AssetSelection.groups("critical"))

If the asset has 10 upstreams but only 2 are in the "critical" group, only those 2 are checked.

ignore() Creates Subtraction

Dependencies in the selection are excluded:

condition = dg.AutomationCondition.any_deps_updated().ignore(
    dg.AssetSelection.assets("test_data", "staging_data")
)

Updates to "test_data" and "staging_data" won't trigger the condition.

Propagation Through Operators

When applied to composite conditions (AND/OR), filtering propagates to all sub-conditions:

# Applies to both any_deps_missing() and any_deps_in_progress() within eager()
condition = dg.AutomationCondition.eager().allow(
    dg.AssetSelection.groups("production")
)

What gets filtered: All any_deps_match() and all_deps_match() calls

What doesn't get filtered: Direct operands like missing() on the asset itself

Understanding since_last_handled()

since_last_handled() is a convenience method that converts events to a status:

condition = dg.AutomationCondition.newly_missing()

# These are equivalent:
condition.since_last_handled()

condition.since(
    dg.AutomationCondition.newly_requested()
    | dg.AutomationCondition.newly_updated()
    | dg.AutomationCondition.initial_evaluation()
)

Behavior:

• Becomes true when condition becomes true
• Stays true until the asset is requested, updated, or the condition is first applied
• Resets on initial evaluation to handle condition changes

Use case: Persist an event until it's "handled" by either requesting or materializing the asset. This prevents duplicate requests while ensuring the event isn't lost.

Composite Conditions Deep Dive

any_deps_updated()

dg.AutomationCondition.any_deps_match(
    (dg.AutomationCondition.newly_updated() & ~dg.AutomationCondition.executed_with_root_target())
    | dg.AutomationCondition.will_be_requested()
)

Checks if any dependency has newly updated (excluding same-run updates) OR will be requested this tick.

any_deps_missing()

dg.AutomationCondition.any_deps_match(
    dg.AutomationCondition.missing() & ~dg.AutomationCondition.will_be_requested()
)

Checks if any dependency is missing AND will NOT be requested this tick. Dependencies that will be requested aren't considered blocking.

all_deps_updated_since_cron()

```python nocheckundefined dg.AutomationCondition.all_deps_match( dg.AutomationCondition.newly_updated().since( dg.AutomationCondition.cron_tick_passed(cron_schedule, cron_timezone) ) )

For each dependency, checks if it has been updated since the last cron tick. All dependencies must have at least one partition updated since the tick.

Declarative Automation: Core Concepts

For basic examples, see the main SKILL.md Quick Reference section on Declarative Automation.

The Three Main Conditions

Dagster provides three primary conditions optimized for common use cases. Start with one of these rather than building conditions from scratch.

eager()

Executes an asset whenever any dependency updates. Also materializes partitions that become missing after the condition is applied.

import dagster as dg

@dg.asset(automation_condition=dg.AutomationCondition.eager())
def downstream_asset(upstream_asset):
    # Executes immediately when upstream_asset materializes
    ...

Behavior:

• Triggers immediately when any upstream updates
• Waits for all upstreams to be materialized or in-progress
• Does not execute if any dependencies are missing
• Does not execute if any dependencies are currently in-progress (waits for all deps to finish first)
• Does not execute if the asset is already in-progress
• For time-partitioned assets, only considers the latest partition
• For static/dynamic-partitioned assets, considers all partitions

Full expanded form:

(
    dg.AutomationCondition.in_latest_time_window()          # latest partition only (time-partitioned)
    & (
        dg.AutomationCondition.newly_missing()
        | dg.AutomationCondition.any_deps_updated()
    ).since_last_handled()                                  # trigger event, persisted until handled
    & ~dg.AutomationCondition.any_deps_missing()            # no deps missing
    & ~dg.AutomationCondition.any_deps_in_progress()        # no deps currently running
    & ~dg.AutomationCondition.in_progress()                 # asset itself not running
).with_label("eager")

The ~any_deps_in_progress() guard is critical: it prevents the asset from firing until ALL upstream deps have finished materializing. Without it, the asset would fire each time an individual dep completes, causing redundant executions when multiple deps update in quick succession (e.g., from the same scheduled job).

Use when: You want updates to propagate downstream immediately without waiting for a schedule.

on_cron()

Executes an asset on a cron schedule after all dependencies have updated since the latest cron tick.

@dg.asset(
    automation_condition=dg.AutomationCondition.on_cron("0 9 * * *", "America/Los_Angeles")
)
def daily_summary(hourly_data):
    # Executes at 9 AM only if hourly_data has updated since the previous 9 AM tick
    ...

Behavior:

• Waits for a cron tick to occur
• After the tick, waits for all dependencies to update since that tick
• Once all dependencies are updated, executes immediately
• For time-partitioned assets, only considers the latest partition

Full expanded form:

cron_schedule = "0 1 * * *"
cron_timezone = "US/Eastern"
(
    dg.AutomationCondition.in_latest_time_window()
    & dg.AutomationCondition.cron_tick_passed(
        cron_schedule, cron_timezone
    ).since_last_handled()
    & dg.AutomationCondition.all_deps_updated_since_cron(cron_schedule, cron_timezone)
).with_label(f"on_cron({cron_schedule}, {cron_timezone})")

Use when: You want scheduled execution but only after upstream data is ready. More intelligent than simple schedules.

on_missing()

Executes missing asset partitions when all upstream partitions are available.

@dg.asset(automation_condition=dg.AutomationCondition.on_missing())
def backfill_asset(upstream):
    # Executes for any missing partitions when upstream is ready
    ...

Behavior:

• Only materializes partitions that are missing
• Only considers partitions added after the condition was applied (not historical)
• Waits for all upstream dependencies to be available
• For time-partitioned assets, only considers the latest partition

Full expanded form:

(
    dg.AutomationCondition.in_latest_time_window()
    & (
        dg.AutomationCondition.missing()
        .newly_true()
        .since_last_handled()
        .with_label("missing_since_last_handled")
    )
    & ~dg.AutomationCondition.any_deps_missing()
).with_label("on_missing")

Use when: You want to fill in missing partitions as upstream data becomes available. Good for backfilling or catching up.

Identifying Built-in vs Custom Conditions from the API

When debugging DA behavior via dg api asset get, the automation_condition.expanded_label field shows the condition tree as a list of strings. Compare this against the full expanded forms above to determine if the asset is using a built-in condition or a custom one with missing guards. When you see a condition that looks similar to but doesn't match a built-in, always identify the missing sub-conditions and explain how their absence changes behavior.

Evaluation by Sensor

The AutomationConditionSensorDefinition evaluates conditions at regular intervals.

Default sensor: A sensor named default_automation_condition_sensor is created automatically in code locations with automation conditions.

Configuration:

• Evaluates all conditions every 30 seconds
• Must be toggled on in the UI under Automation → Sensors
• Launches runs when conditions evaluate to true

Important: If the sensor is not enabled, conditions will not be evaluated and no runs will launch.

Basic Customization

All three main conditions are built from smaller components and can be customized.

Modifying conditions

# Remove sub-conditions
condition = dg.AutomationCondition.eager().without(
    ~dg.AutomationCondition.any_deps_missing()
)

# Replace sub-conditions
condition = dg.AutomationCondition.on_cron("0 9 * * *").replace(
    old=dg.AutomationCondition.all_deps_updated_since_cron("0 9 * * *"),
    new=dg.AutomationCondition.all_deps_updated_since_cron("0 0 * * *"),
)

Boolean composition

# AND: Both conditions must be true
condition = (
    dg.AutomationCondition.eager()
    & ~dg.AutomationCondition.in_progress()
)

# OR: Either condition can be true
condition = (
    dg.AutomationCondition.on_cron("0 9 * * *")
    | dg.AutomationCondition.any_deps_updated()
)

See customization.md for detailed patterns and examples.

When to Use Declarative Automation

Use declarative automation when:

• Asset-centric pipelines with complex update logic
• Condition-based triggers (data availability, freshness)
• Dependency-aware execution is needed
• You prefer declarative over imperative

Use schedules when:

• Simple time-based execution without dependency logic
• Predictable, fixed-time execution is sufficient

Use sensors when:

• Custom polling logic for external systems
• Imperative actions beyond asset execution
• File watching or API event monitoring

Declarative Automation: Customization

Start with one of the three main conditions (eager(), on_cron(), on_missing()) and customize them using these patterns.

Pattern 1: Removing Sub-conditions with without()

Remove unwanted sub-conditions from composite conditions.

Allow missing upstreams: By default, eager() waits for all dependencies. Remove this requirement:

import dagster as dg

condition = (
    dg.AutomationCondition.eager()
    .without(~dg.AutomationCondition.any_deps_missing())
    .with_label("eager_allow_missing")
)

Update all time partitions: By default, eager() only updates the latest time partition. Remove this restriction:

condition = (
    dg.AutomationCondition.eager()
    .without(dg.AutomationCondition.in_latest_time_window())
    .with_label("eager_all_partitions")
)

Pattern 2: Replacing Sub-conditions with replace()

Swap one sub-condition for another with different parameters.

Multiple cron schedules: Execute at 9 AM but wait for dependencies to update since midnight:

NINE_AM_CRON = "0 9 * * *"
MIDNIGHT_CRON = "0 0 * * *"

condition = dg.AutomationCondition.on_cron(NINE_AM_CRON).replace(
    old=dg.AutomationCondition.all_deps_updated_since_cron(NINE_AM_CRON),
    new=dg.AutomationCondition.all_deps_updated_since_cron(MIDNIGHT_CRON),
)

Partition lookback window: Expand on_missing() to consider the last 24 hours of partitions:

import datetime

condition = dg.AutomationCondition.on_missing().replace(
    old=dg.AutomationCondition.in_latest_time_window(),
    new=dg.AutomationCondition.in_latest_time_window(
        lookback_delta=datetime.timedelta(hours=24)
    ),
)

Pattern 3: Filtering Dependencies with allow() and ignore()

Control which dependencies are considered.

Only specific dependencies: Only trigger on updates from assets in the "abc" group:

condition = dg.AutomationCondition.eager().allow(
    dg.AssetSelection.groups("abc")
)

Exclude specific dependencies: Ignore updates from the "foo" asset:

condition = dg.AutomationCondition.eager().ignore(
    dg.AssetSelection.assets("foo")
)

Pattern 4: Boolean Composition

Combine multiple conditions with AND (&), OR (|), NOT (~).

Scheduled with dependency-driven fallback: Run every 5 minutes or when dependencies update (if updated today):

daily_success_condition = dg.AutomationCondition.newly_updated().since(
    dg.AutomationCondition.cron_tick_passed("0 0 * * *")
)

condition = (
    dg.AutomationCondition.cron_tick_passed("*/5 * * * *")
    | (
        dg.AutomationCondition.any_deps_updated()
        & daily_success_condition
        & ~dg.AutomationCondition.any_deps_missing()
        & ~dg.AutomationCondition.any_deps_in_progress()
    )
)

Only execute when checks pass: Ensure all blocking checks on dependencies pass:

condition = (
    dg.AutomationCondition.eager()
    & dg.AutomationCondition.all_deps_match(
        dg.AutomationCondition.all_checks_match(
            dg.AutomationCondition.check_passed(),
            blocking_only=True,
        )
    )
)

Pattern 5: Custom Event-Based Conditions

Build conditions from operands and operators for specific scenarios.

On code version change: Execute when code version changes:

condition = (
    dg.AutomationCondition.code_version_changed().since_last_handled()
    & ~dg.AutomationCondition.any_deps_missing()
)

After upstream success: Execute only after a specific upstream asset updates:

condition = (
    dg.AutomationCondition.any_deps_match(
        dg.AutomationCondition.newly_updated()
    ).allow(dg.AssetSelection.assets("critical_upstream"))
    .since_last_handled()
)

Combining Patterns

Multiple patterns can be combined for complex requirements:

condition = (
    dg.AutomationCondition.eager()
    .without(dg.AutomationCondition.in_latest_time_window())  # Pattern 1
    .ignore(dg.AssetSelection.assets("staging_data"))         # Pattern 3
    & dg.AutomationCondition.all_checks_match(                # Pattern 4
        dg.AutomationCondition.check_passed(),
        blocking_only=True,
    )
).with_label("custom_backfill_with_checks")

This condition uses eager() as the base, removes the latest partition restriction, ignores a specific dependency, and adds a check requirement.

Declarative Automation Reference

Declarative automation uses AutomationCondition objects to describe when assets should execute. Instead of scheduling jobs, you define conditions on assets that the system evaluates automatically.

Overview

Modern automation pattern: Set conditions directly on assets rather than creating separate schedules or sensors. The system evaluates conditions every 30 seconds and launches runs when conditions are met.

Benefits:

• Asset-native: No separate job definitions needed
• Dependency-aware: Automatically considers upstream state
• Composable: Build complex conditions from simple building blocks
• Declarative: Easier to reason about than imperative sensors

Basic examples: See the main SKILL.md Quick Reference for eager(), on_cron(), and on_missing() examples.

Requirements

• Assets only: Declarative automation does not work with ops or graphs
• Sensor must be enabled: The default_automation_condition_sensor must be toggled on in the Dagster UI under Automation → Sensors

Core Concepts

The Three Main Conditions

Start with one of these three conditions rather than building conditions from scratch:

• `eager()`: Execute immediately when dependencies update
• `on_cron()`: Execute on a schedule after dependencies update
• `on_missing()`: Execute missing partitions when dependencies are ready

Customization

All three main conditions can be customized:

• Remove sub-conditions with .without()
• Replace sub-conditions with .replace()
• Filter dependencies with .allow() and .ignore()
• Combine with boolean operators: & (AND), | (OR), ~ (NOT)

Advanced Concepts

• Status vs Events: Conditions can be persistent states or transient moments
• Operands: Base building blocks like missing(), newly_updated()
• Operators: Tools for composition like since(), any_deps_match()

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• Advanced — status vs events, run grouping, or filtering in declarative automation
• Core Concepts — using eager(), on_cron(), or on_missing() conditions
• Customization — customizing conditions with without(), replace(), allow(), or ignore()
• Operands — base condition building blocks like missing() or newly_updated()
• Operators — combining conditions using since, any_deps_match, or boolean operators

<!-- END GENERATED INDEX -->

Declarative Automation: Operands

Operands are base conditions that evaluate to true or false for a given asset or asset partition. They represent fundamental states and events.

Complete List of Operands

Status vs Event Operands

Status operands are persistent and remain true for multiple evaluations:

• missing() - Stays true until the asset is materialized
• in_progress() - True while a run is executing
• execution_failed() - True until the asset succeeds or is re-requested
• in_latest_time_window() - True for the latest time partition(s)
• will_be_requested() - True during the tick when a request will be made

Event operands are transient and true for only one evaluation:

• newly_updated() - True only on the tick when the update occurs
• newly_requested() - True only on the tick when the request is made
• code_version_changed() - True only on the first tick after the change
• cron_tick_passed() - True only on the first tick after the cron tick
• initial_evaluation() - True only on the very first evaluation

Detailed Descriptions

`missing()`: True if the asset partition has never been materialized or observed.

`in_progress()`: True if the asset partition is part of an in-progress run or backfill. Combines run_in_progress() and backfill_in_progress().

`execution_failed()`: True if the latest execution of the asset partition failed.

`newly_updated()`: True if the asset partition was materialized or observed since the previous evaluation. For observations, only true if the data version changed.

`newly_requested()`: True if the asset partition was requested on the previous evaluation tick.

`code_version_changed()`: True if the asset's code version has changed since the previous evaluation.

`cron_tick_passed(cron_schedule, cron_timezone)`: True on the first evaluation after a tick of the specified cron schedule occurs.

Parameters:

• cron_schedule (str): Cron expression
• cron_timezone (str): Timezone string (default: "UTC")

`in_latest_time_window(lookback_delta)`: True for time partitions within the latest time window. For unpartitioned or non-time-partitioned assets, always true.

Parameter:

• lookback_delta (Optional[timedelta]): If provided, returns partitions within this delta of the latest window end. For daily partitions with lookback_delta=timedelta(hours=48), returns the latest 2 partitions.

`will_be_requested()`: True if the asset partition will be requested in the current tick. Used internally for run grouping (see advanced.md).

`initial_evaluation()`: True only on the first evaluation after the condition is applied or modified.

Composite Conditions

Built from base operands for convenience:

Usage

Operands are composed using operators (see operators.md) to build complex conditions:

import dagster as dg

# Using operands directly
condition = dg.AutomationCondition.missing() & ~dg.AutomationCondition.in_progress()

# Using composite conditions
condition = dg.AutomationCondition.any_deps_updated()

Declarative Automation: Operators

Operators combine operands and other conditions into complex expressions using boolean logic and transformations.

Boolean Operators

`&` (AND): Both conditions must be true:

import dagster as dg

condition = (
    dg.AutomationCondition.newly_updated()
    & ~dg.AutomationCondition.in_progress()
)

`|` (OR): Either condition must be true:

condition = (
    dg.AutomationCondition.missing()
    | dg.AutomationCondition.newly_updated()
)

`~` (NOT): Negates the condition:

condition = ~dg.AutomationCondition.any_deps_missing()

Transformation Operators

since(reset_condition)

Converts events into status. Becomes true when the operand becomes true and remains true until the reset condition becomes true.

# True from when dependency updates until asset is requested
condition = dg.AutomationCondition.any_deps_updated().since(
    dg.AutomationCondition.newly_requested()
)

Pattern: A.since(B) means "A has occurred more recently than B"

Use case: Create persistent states from transient events. "Upstream updated" is an event, but "upstream updated since I was last requested" is a status.

newly_true()

Converts status into an event. True only on the tick when the operand transitions from false to true.

# True only on the tick when the asset becomes missing
condition = dg.AutomationCondition.missing().newly_true()

Use case: Prevent repeated actions during persistent states. missing() stays true for many ticks, but missing().newly_true() is only true once.

since_last_handled()

Convenience method equivalent to .since(newly_requested() | newly_updated() | initial_evaluation()).

condition = dg.AutomationCondition.any_deps_updated().since_last_handled()

True from when the condition becomes true until the asset is requested, updated, or the condition is first applied.

Dependency Operators

any_deps_match(condition)

True if the condition is true for at least one partition of any upstream dependency.

condition = dg.AutomationCondition.any_deps_match(
    dg.AutomationCondition.missing()
)

Supports filtering with .allow() and .ignore().

all_deps_match(condition)

True if the condition is true for at least one partition of all upstream dependencies.

condition = dg.AutomationCondition.all_deps_match(
    dg.AutomationCondition.newly_updated()
)

Requires every upstream asset to have at least one partition matching the condition.

Dependency Filtering

allow(selection)

Restricts which dependencies are checked to only those in the AssetSelection:

# Only consider dependencies in the "important" group
condition = dg.AutomationCondition.any_deps_match(
    dg.AutomationCondition.missing()
).allow(dg.AssetSelection.groups("important"))

Creates an intersection: dep_keys & allowed_selection

ignore(selection)

Excludes dependencies in the AssetSelection from being checked:

# Ignore the "foo" asset when checking for updates
condition = dg.AutomationCondition.any_deps_updated().ignore(
    dg.AssetSelection.assets("foo")
)

Creates a subtraction: dep_keys - ignored_selection

Propagation Through Boolean Operators

When applied to AND/OR conditions, .allow() and .ignore() propagate to all sub-conditions:

# Applies allow() to all dependency checks within eager()
condition = dg.AutomationCondition.eager().allow(
    dg.AssetSelection.groups("critical")
)

Check Operators

any_checks_match(condition, blocking_only)

True if any of the asset's checks match the condition.

condition = dg.AutomationCondition.any_checks_match(
    dg.AutomationCondition.check_failed(),
    blocking_only=True,
)

Parameters:

• condition: Condition to evaluate against checks
• blocking_only (bool): If True, only considers blocking checks (default: False)

all_checks_match(condition, blocking_only)

True if all of the asset's checks match the condition.

condition = dg.AutomationCondition.all_checks_match(
    dg.AutomationCondition.check_passed(),
    blocking_only=True,
)

Labeling

with_label(label)

Adds a human-readable label to a condition for debugging and UI display:

condition = (
    dg.AutomationCondition.any_deps_updated()
    .since(dg.AutomationCondition.newly_requested())
).with_label("updated_since_requested")

Labels appear in condition evaluation traces in the Dagster UI, making complex conditions easier to understand.

Schedules

Basic schedule patterns are covered in the main SKILL.md Quick Reference. This reference covers advanced schedule configuration and partitioned job automation.

Basic Schedule Review

A schedule executes a job at specified times using cron expressions. See SKILL.md for the basic pattern.

import dagster as dg

daily_job = dg.define_asset_job("daily_job", selection="*")

daily_schedule = dg.ScheduleDefinition(
    job=daily_job,
    cron_schedule="0 0 * * *",  # Midnight UTC
)

Execution Timezone

Schedules default to UTC. Specify a different timezone with execution_timezone:

```python nocheckundefined daily_schedule = dg.ScheduleDefinition( job=daily_refresh_job, cron_schedule="0 9 *", # 9 AM execution_timezone="America/Los_Angeles", )

**Timezone string format**: Use IANA timezone database names like `"America/New_York"`, `"Europe/London"`, or `"Asia/Tokyo"`.

**Daylight saving time**: Dagster handles DST transitions automatically based on the specified timezone.

## Schedules from Partitioned Assets

For partitioned assets or jobs, use `build_schedule_from_partitioned_job` to automatically create a schedule matching the partition cadence:

@dg.asset(partitions_def=dg.DailyPartitionsDefinition(start_date="2024-01-01")) def daily_asset(context: dg.AssetExecutionContext): partition_date = context.partition_key

Process data for this partition

...

partitioned_job = dg.define_asset_job( name="daily_partitioned_job", selection=[daily_asset] )

Schedule automatically inherits daily cadence and timezone from partition definition

schedule = dg.build_schedule_from_partitioned_job(partitioned_job)

**How it works**: The schedule's cron expression is derived from the `PartitionsDefinition`:

- `DailyPartitionsDefinition` → daily cron schedule
- `WeeklyPartitionsDefinition` → weekly cron schedule
- `MonthlyPartitionsDefinition` → monthly cron schedule
- `HourlyPartitionsDefinition` → hourly cron schedule

Each schedule run materializes the partition corresponding to the schedule time.

## Cron Expression Reference

Common cron patterns for schedules:

| Cron Expression  | Description                 |
| ---------------- | --------------------------- |
| `0 * * * *`      | Every hour                  |
| `0 0 * * *`      | Daily at midnight           |
| `0 9 * * *`      | Daily at 9 AM               |
| `0 0 * * 1`      | Weekly on Monday            |
| `0 0 1 * *`      | Monthly on the 1st          |
| `0 0 1 1 *`      | Yearly on January 1st       |
| `*/15 * * * *`   | Every 15 minutes            |
| `0 9-17 * * 1-5` | Hourly, 9 AM-5 PM, weekdays |

**Cron format**: `minute hour day_of_month month day_of_week`

## Configuration Options

schedule = dg.ScheduleDefinition( job=my_job, cron_schedule="0 0 *", execution_timezone="UTC", default_status=dg.DefaultScheduleStatus.RUNNING, # Start enabled description="Daily data refresh for analytics", tags={"team": "data-eng", "priority": "high"}, )

**Key parameters**:

- `default_status`: Set to `DefaultScheduleStatus.RUNNING` to enable schedule automatically when deployed (default is `STOPPED`)
- `description`: Human-readable description shown in the Dagster UI
- `tags`: Metadata tags for organization and filtering

## When to Use Schedules

**Use schedules when**:

- Execution time is predictable and fixed
- No dependency logic is needed (runs regardless of upstream status)
- Simple time-based triggers are sufficient

**Prefer declarative automation when**:

- You need dependency-aware execution
- Conditions involve asset freshness or upstream state
- Complex logic determines when to execute

Asset Sensors

Asset sensors monitor asset materializations and trigger jobs when specific assets are materialized.

Basic Asset Sensor

Use @asset_sensor to monitor a specific asset:

```python nocheckundefined @dg.asset_sensor(asset_key=dg.AssetKey("daily_sales_data"), job=downstream_job) def sales_data_sensor(context: dg.SensorEvaluationContext, asset_event: dg.EventLogEntry):

Triggered whenever daily_sales_data is materialized

yield dg.RunRequest(run_key=context.cursor)

The sensor is called once per materialization event with the event details in `asset_event`.

## Cross-Job Dependencies

Asset sensors enable dependencies across different jobs:

Job A contains upstream_asset

@dg.asset def upstream_asset(): ...

job_a = dg.define_asset_job("job_a", selection=[upstream_asset])

Job B is triggered when upstream_asset materializes

@dg.asset_sensor(asset_key=dg.AssetKey("upstream_asset"), job=job_b) def cross_job_sensor(context, asset_event): return dg.RunRequest()

This pattern is useful when you have logically separate jobs that need coordination.

## Cross-Code Location Dependencies

Asset sensors can monitor assets in different code locations:

@dg.asset_sensor( asset_key=dg.AssetKey("other_location_asset"), job=my_job, ) def cross_location_sensor(context, asset_event): return dg.RunRequest()

The sensor can be in a different code location than the monitored asset.

## Custom Evaluation Logic

Add conditional logic to control when to trigger:

@dg.asset_sensor(asset_key=dg.AssetKey("daily_sales_data"), job=downstream_job) def conditional_sensor(context, asset_event):

Access materialization metadata

metadata = asset_event.dagster_event.event_specific_data.materialization.metadata

row_count = metadata.get("row_count").value if "row_count" in metadata else 0

if row_count > 1000: return dg.RunRequest() else: return dg.SkipReason(f"Row count {row_count} too low, threshold is 1000")

**Use cases for conditional logic**: Trigger downstream processing only when data volume is sufficient, quality checks pass, or specific metadata conditions are met.

## Triggering with Custom Configuration

Pass runtime configuration to the triggered job:

@dg.asset_sensor(asset_key=dg.AssetKey("source_data"), job=processing_job) def config_sensor(context, asset_event):

Extract partition key from the materialization

partition_key = asset_event.dagster_event.logging_tags.get("dagster/partition")

return dg.RunRequest( run_key=partition_key, tags={"dagster/partition": partition_key}, )

This allows you to propagate partition information or other metadata from the upstream asset to the triggered job.

## Asset Sensors vs Declarative Automation

**Use asset sensors when**:

- Triggering imperative side effects (notifications, external API calls)
- Launching jobs with complex custom logic
- Cross-code location dependencies with conditional triggers
- Need to inspect materialization metadata before deciding to trigger

**Use declarative automation when**:

- Automating asset-to-asset execution within the same code location
- Defining dependencies based on asset freshness or missing status
- Requiring sophisticated dependency logic with composable conditions

Declarative automation is the recommended approach for asset-centric workflows. Asset sensors remain valuable for triggering actions outside the asset graph or when you need imperative control.

Basic Sensors

For the basic sensor pattern with cursors, see the main SKILL.md Quick Reference section.

File Watching Sensor

A canonical file sensor that monitors a directory for new files and triggers runs:

```python nocheckundefined import os import json import dagster as dg

@dg.sensor(job=my_job, minimum_interval_seconds=30) def file_sensor(context: dg.SensorEvaluationContext):

Load cursor (tracks files we've already processed)

processed_files = json.loads(context.cursor) if context.cursor else {}

Check directory for files

directory = "/data/incoming" current_files = {} runs_to_request = []

for filename in os.listdir(directory): filepath = os.path.join(directory, filename) mtime = os.path.getmtime(filepath) current_files[filename] = mtime

File is new or modified

if filename not in processed_files or processed_files[filename] != mtime: runs_to_request.append( dg.RunRequest( run_key=f"{filename}_{mtime}", run_config={"ops": {"my_op": {"config": {"filepath": filepath}}}}, ) )

Update cursor to track current state

return dg.SensorResult( run_requests=runs_to_request, cursor=json.dumps(current_files), )

**Key pattern**: Store file names and modification times in the cursor to track which files have been processed.

## Cursor State Management

**Best practices for cursors**:

- **Use JSON for structured state**: `json.dumps()` and `json.loads()` make it easy to store dictionaries or lists
- **Handle missing cursor**: Always check if `context.cursor` is None on first evaluation
- **Update cursor atomically**: Return the new cursor value in `SensorResult` or call `context.update_cursor()`
- **Keep cursors small**: Cursors are stored in the database; avoid storing large data structures

**Two ways to update cursors**:

Option 1: Return SensorResult

return dg.SensorResult( run_requests=[...], cursor=json.dumps(new_state) )

Option 2: Call update_cursor() directly

context.update_cursor(json.dumps(new_state)) yield dg.RunRequest(...)

## Evaluation Configuration

**Control evaluation frequency**:

@dg.sensor( job=my_job, minimum_interval_seconds=60, # Minimum 60 seconds between evaluations default_status=dg.DefaultSensorStatus.RUNNING, # Auto-enable when deployed ) def my_sensor(context): ...

**Important**: `minimum_interval_seconds` is a minimum, not exact. If sensor evaluation takes 10 seconds and the interval is 30 seconds, the next evaluation happens 30 seconds after the previous evaluation started (20 seconds after it completed).

## SensorEvaluationContext

Properties available in sensor context:

- `cursor`: String cursor from the previous evaluation (None if first evaluation)
- `update_cursor(str)`: Update the cursor for the next evaluation
- `instance`: DagsterInstance for querying the event log or other instance data
- `log`: Logger for recording sensor evaluation details
- `repository_def`: Repository containing the sensor
- `resources`: Access configured resources (if defined)

**Example using context.log**:

@dg.sensor(job=my_job) def logging_sensor(context): context.log.info(f"Evaluating sensor, cursor: {context.cursor}")

... sensor logic

Run Status Sensors

Run status sensors monitor runs for specific status changes and trigger actions when those statuses occur.

Run Failure Sensor

Use @run_failure_sensor to monitor run failures across all jobs:

```python nocheckundefined import dagster as dg

@dg.run_failure_sensor def failure_alert_sensor(context: dg.RunFailureSensorContext): slack_client.chat_postMessage( channel="#alerts", text=f'Job "{context.dagster_run.job_name}" failed: {context.failure_event.message}', )

Run failure sensors are commonly used for alerting and error notification.

## Run Status Sensor

Use `@run_status_sensor` to monitor any run status:

@dg.run_status_sensor( run_status=dg.DagsterRunStatus.SUCCESS, request_job=downstream_job, ) def success_sensor(context: dg.RunStatusSensorContext): if context.dagster_run.job_name == "upstream_job": return dg.RunRequest(run_key=context.dagster_run.run_id)

This pattern enables job-to-job dependencies based on run completion.

## Available Run Statuses

Common run statuses for monitoring:

- `DagsterRunStatus.SUCCESS` - Run completed successfully
- `DagsterRunStatus.FAILURE` - Run failed
- `DagsterRunStatus.STARTED` - Run execution started
- `DagsterRunStatus.CANCELED` - Run was canceled
- `DagsterRunStatus.CANCELING` - Run is being canceled

Additional statuses: `QUEUED`, `NOT_STARTED`, `MANAGED`, `STARTING`

## Monitoring Specific Jobs

Use `monitored_jobs` to filter which jobs trigger the sensor:

@dg.run_status_sensor( run_status=dg.DagsterRunStatus.SUCCESS, monitored_jobs=[job1, job2], ) def job_specific_sensor(context):

Only triggered when job1 or job2 succeeds

...

Without `monitored_jobs`, the sensor triggers for all runs with the specified status.

## Cross-Code Location Monitoring

Monitor runs across all code locations:

@dg.run_status_sensor( run_status=dg.DagsterRunStatus.FAILURE, monitor_all_code_locations=True, ) def global_failure_sensor(context):

Monitors failures across the entire deployment

...

Set `monitor_all_code_locations=True` to enable deployment-wide monitoring.

## Context Properties

**RunStatusSensorContext** provides:

- `dagster_run`: The run that triggered the sensor
- `dagster_event`: The event associated with the status change
- `partition_key`: Partition key from run tags (if partitioned)
- `instance`: DagsterInstance
- `log`: Logger

**RunFailureSensorContext** adds:

- `failure_event`: The run failure event with error details
- `get_step_failure_events()`: List of step-level failures with stack traces

## Common Use Cases

**Alerting**: Send notifications on run failures:

@dg.run_failure_sensor def slack_alert(context: dg.RunFailureSensorContext): slack_client.chat_postMessage( channel="#alerts", text=f"Job {context.dagster_run.job_name} failed" )

**Job coordination**: Trigger downstream jobs after success:

@dg.run_status_sensor( run_status=dg.DagsterRunStatus.SUCCESS, monitored_jobs=[upstream_job], request_job=downstream_job, ) def chain_jobs(context): return dg.RunRequest()

**Error handling**: Trigger cleanup on failure:

@dg.run_failure_sensor(monitored_jobs=[data_job]) def cleanup_on_failure(context): cleanup_partial_data()

dg api agent get <AGENT_ID>

dg api agent Reference

Commands for listing and inspecting Dagster Plus agents.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api agent get — details about a specific Dagster Plus agent
• dg api agent list — listing agents in Dagster Plus

<!-- END GENERATED INDEX -->

dg api agent list

dg api alert-policy Reference

Commands for managing alert policies in Dagster Plus.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api alert-policy list — listing alert policies in Dagster Plus
• dg api alert-policy sync — syncing alert policies from YAML definition

<!-- END GENERATED INDEX -->

List alert policies for a deployment.

dg api alert-policy list

Sync alert policies from a YAML definition file. This will create, update, or remove alert policies to match the definition file.

dg api alert-policy sync <FILE>

• <FILE> — path to a YAML file defining the desired alert policies

Download an artifact by key from Dagster Plus.

dg api artifact download <KEY> <OUTPUT_PATH>

• <KEY> — the artifact key to download
• <OUTPUT_PATH> — local path to write the downloaded file

dg api artifact Reference

Commands for uploading and downloading artifacts in Dagster Plus.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api artifact download — downloading an artifact from Dagster Plus
• dg api artifact upload — uploading an artifact to Dagster Plus

<!-- END GENERATED INDEX -->

Upload an artifact by key to Dagster Plus.

dg api artifact upload <KEY> <FILE>

• <KEY> — the artifact key to upload under
• <FILE> — path to the file to upload

dg api asset-check Reference

Commands for querying asset checks in a Dagster Plus deployment.

dg api asset-check list

dg api asset-check list --asset-key <ASSET_KEY>

• --asset-key (required) — slash-separated asset key (e.g. my/asset)

dg api asset-check get-executions

dg api asset-check get-executions --asset-key <ASSET_KEY> --check-name <CHECK_NAME>

• --asset-key (required) — slash-separated asset key (e.g. my/asset)
• --check-name (required) — name of the asset check
• --limit — max results (default: 25)
• --cursor — pagination cursor

dg api asset get-evaluations <ASSET_KEY>

--include-nodes — includes individual evaluation nodes in the response. Warning: this produces dense output with the full tree of conditions evaluated for each record. Only use when processing a small number of records.

dg api asset get-events <ASSET_KEY>

• --event-type — filter by event type (e.g. ASSET_MATERIALIZATION, ASSET_OBSERVATION)
• --partition — filter events by partition key
• --before — return events before this timestamp; use with --limit to paginate chronologically

Get asset health and runtime status information.

dg api asset get-health <ASSET_KEY>

For prefixed asset keys, use slash-separated syntax: dg api asset get-health my_prefix/my_asset.

dg api asset get-partition-status <ASSET_KEY>

Returns partition materialization statistics (materialized, failed, missing counts) for a partitioned asset.

dg api asset get <ASSET_KEY>

For prefixed asset keys, use slash-separated syntax: dg api asset get my_prefix/my_asset.

--status — includes materialization status information. Not included by default as it requires additional API calls.

dg api asset Reference

Commands for querying information about assets in a Dagster Plus deployment.

To materialize an asset on a deployed Dagster Plus environment, see `dg api run launch`. For local in-process materialization, see `dg launch`.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api asset get-evaluations — automation condition evaluation history for an asset
• dg api asset get-events — materialization or observation event history for an asset
• dg api asset get-health — getting asset health or runtime status
• dg api asset get-partition-status — partition materialization status or stats for an asset
• dg api asset get — details about a specific asset
• dg api asset list — querying which assets exist in a deployment

<!-- END GENERATED INDEX -->

dg api asset list

• --status — includes detailed status information in the response
• --limit / --cursor — pagination support

Add or update a code location in a deployment.

dg api code-location add <NAME>

• --image — container image to use
• --module — Python module to load definitions from
• --package — Python package to load definitions from
• --python-file — Python file to load definitions from

Delete a code location from a deployment.

dg api code-location delete <NAME>

Get details for a specific code location.

dg api code-location get <NAME>

dg api code-location Reference

Commands for managing code locations in a Dagster Plus deployment.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api code-location add — adding or updating a code location in Dagster Plus
• dg api code-location delete — deleting a code location from Dagster Plus
• dg api code-location get — details about a specific code location
• dg api code-location list — listing code locations in Dagster Plus

<!-- END GENERATED INDEX -->

List code locations in a deployment.

dg api code-location list

Delete a deployment.

dg api deployment delete <NAME>

dg api deployment get <NAME>

dg api deployment Reference

Commands for managing Dagster Plus deployments and their settings.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api deployment delete — deleting a deployment from Dagster Plus
• dg api deployment get — details about a specific deployment
• dg api deployment list — listing deployments in Dagster Plus
• dg api deployment settings-get — getting deployment-level settings in Dagster Plus
• dg api deployment settings-set — setting deployment settings in Dagster Plus

<!-- END GENERATED INDEX -->

dg api deployment list

Get deployment-level settings.

dg api deployment settings-get

Set deployment settings from a YAML file.

dg api deployment settings-set <FILE>

• <FILE> — path to a YAML file defining the desired deployment settings

All dg api subcommands support --json, --response-schema, --deployment, --organization, --api-token, and --view-graphql.

• --response-schema — prints the JSON schema for the command's response and exits. Run this before writing any parsing logic to get exact field names, types, and valid enum values.
• --view-graphql — prints GraphQL queries and responses to stderr, useful for debugging.

Tips

For complex debugging/analysis workflows, ALWAYS use --json to get machine-readable output. Pipe into jq (recommended) or other tools for further processing.

Flags like --deployment/--organization/--api-token are typically not needed when authenticated via dg plus login.

Dagster Plus API Reference

The dg api subcommands provide CLI access to Dagster Plus resources. They are useful for scripting, debugging, and automation workflows.

Important: Always read general.md first before using any dg api subcommand. It covers shared flags (--json, --response-schema, --deployment, --view-graphql) and best practices.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api asset-check — querying asset checks or asset check execution history in Dagster Plus
• dg api: General — always read before using any dg api subcommand
• dg api job — listing or inspecting jobs in Dagster Plus
• dg api agent — listing or inspecting Dagster Plus agents
• dg api alert-policy — managing alert policies in Dagster Plus (listing, syncing from YAML)
• dg api artifact — uploading or downloading artifacts in Dagster Plus
• dg api asset — querying information about assets in Dagster Plus (metadata, events, DA evaluations, health status, etc.)
• dg api code-location — managing code locations in Dagster Plus (add, delete, list, inspect)
• dg api deployment — managing Dagster Plus deployments (create, delete, list, settings)
• dg api issue — listing Dagster Plus issues, fetching a specifc Dagster Plus issue
• dg api organization — managing Dagster Plus organization settings and SAML/SSO configuration
• dg api run — run operations, run details, run events, run logs; listing runs, getting run info, fetching run events, compute logs
• dg api schedule — schedule operations, schedule details, schedule ticks; listing schedules, getting schedule info, schedule tick history
• dg api secret — listing or inspecting secrets in a Dagster Plus deployment
• dg api sensor — sensor operations, sensor details, sensor ticks; listing sensors, getting sensor info, sensor tick history

<!-- END GENERATED INDEX -->

dg api issue Reference

Commands for interacting with Dagster Plus Issues.

A Dagster Plus Issue is a record of a problem within the Users' Dagster deployment like you would find in an issue tracking tool. Issues have the following fields: ID, title, description, status, createdBy, links to related Runs and Assets, and additional context about the Issue, including any previous conversations about the problem.

Some organizations do not have access to Dagster Plus Issues. If you get an Unauthorized error indicating that Issues are not available, inform the user that Issues are not enabled for their organization.

Get a specific Dagster Plus Issue

dg api issue get <ID>

• <ID> — the ID of the Issue to get

List Issues for a deployment.

dg api issue list

Issues can be filtered by:

• Status: --status - options are OPEN, CLOSED, TRIAGE. Multiple --status filters can be specified
• Created before: --created-before - filter to Issues created before this date
• Created after: --created-after - filter to Issues created after this date

The response will contain a list of limit Issues in chronologically descending order. To fetch the next page of Issues, use the ID of the oldest Issue as the cursor.

Create a Dagster Plus Issue

dg api issue create --title <title> --description <description> --status <status>

• <title> - The title should be short and clearly state the problem to fix so that the reader quickly understands the cause of the problem. Do not mention specific run ids or other downstream impacts.
• <description> - The description should be a total of 2-4 bullet points that outline the root cause of the problem and next steps.
• --status (optional) - updates the status of the Issue. One of OPEN, CLOSED, TRIAGE, CANCELED

Update a Dagster Plus Issue

dg api issue update <ID>

• <ID> - The ID of the Issue to update
• --status (optional) - updates the status of the Issue. One of OPEN, CLOSED, TRIAGE, CANCELED
• --title (optional) - updates the title of the Issue
• --description (optional) - updates the description of the Issue
• --context (optional) - replaces the additional context stored about this Issue. If you want to append to the current context, fetch the Issue first, append to the context string with the new information, then call the update command with the resulting context.

Link a run or asset to an Issue

 dg api issue add-link <ID>

• <ID> - The ID of the Issue
• --run-id (optional) - The run id of the run to link to the Issue
• --asset-key (optional) - The asset key of the asset to link to the Issue. The asset key should be slash-separated (e.g. my/asset)

Remove a linked run or asset from an Issue

 dg api issue remove-link <ID>

• <ID> - The ID of the Issue
• --run-id (optional) - The run id of the run to remove from the Issue
• --asset-key (optional) - The asset key of the asset to remove from to the Issue. The asset key should be slash-separated (e.g. my/asset)

dg api job Reference

Commands for querying jobs in a Dagster Plus deployment.

dg api job list

dg api job list

Lists all jobs in the deployment.

dg api job get

dg api job get <JOB_NAME>

Returns details for a specific job in the deployment.

Launching jobs

To launch a job on a deployed Dagster Plus environment, see `dg api run launch`. For local in-process execution during development, see `dg launch`.

dg api organization Reference

Commands for managing Dagster Plus organization settings and SAML/SSO configuration.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api organization saml-remove — removing SAML metadata from Dagster Plus
• dg api organization saml-upload — uploading SAML metadata for SSO in Dagster Plus
• dg api organization settings-get — getting organization-level settings in Dagster Plus
• dg api organization settings-set — setting organization settings in Dagster Plus

<!-- END GENERATED INDEX -->

Remove SAML metadata from the organization, disabling SSO.

dg api organization saml-remove

Upload SAML metadata for SSO configuration.

dg api organization saml-upload <FILE>

• <FILE> — path to the SAML metadata XML file

Get organization-level settings.

dg api organization settings-get

Set organization settings from a YAML file.

dg api organization settings-set <FILE>

• <FILE> — path to a YAML file defining the desired organization settings

dg api run get-events <RUN_ID>

• --level — filter by log level: DEBUG, INFO, WARNING, ERROR, CRITICAL. Repeatable.
• --event-type — filter by event type (e.g. STEP_FAILURE, RUN_START). Repeatable.
• --step — filter by step key. Supports partial matching — --step my_asset will match step keys containing that substring. Repeatable.
• --limit — maximum number of events to return.
• --cursor — pagination cursor for retrieving more events.

dg api run get-logs <RUN_ID>

• --step-key — filter to a specific step.
• --link-only — return download URLs instead of log content.
• --max-bytes — maximum bytes of log content per step.
• --cursor — cursor for paginating log content.
• --json — output in JSON format.

dg api run get <RUN_ID>

Run API Commands

Commands for interacting with Dagster runs via dg api run.

<!-- BEGIN GENERATED INDEX -->

• dg api run get-events — debugging a run by reading its logs; filtering run events by level or step
• dg api run get-logs — fetching stdout stderr compute logs for a run; downloading step output logs
• dg api run get — details about a specific run
• dg api run launch — materializing assets or launching jobs on a Dagster Plus deployment
• dg api run list — listing or filtering runs

<!-- END GENERATED INDEX -->

dg api run launch launches a run on a remote Dagster Plus deployment. Use this for materializing assets or launching jobs against your deployed environment. For local in-process execution during development, use `dg launch` instead.

dg api run launch --location <LOCATION> --job <JOB_NAME>
dg api run launch --location <LOCATION> --asset-key <KEY> [--asset-key <KEY> ...]

• --location / -l (required) — code location name
• --repository / -r — repository name (default: __repository__)
• --job / -j — name of the job to launch
• --asset-key — asset key to materialize. Repeatable. Use slash-separated syntax for prefixed keys (e.g. my_prefix/my_asset). The asset selection DSL (group:, tag:, +upstream) is not supported here — list explicit keys only. For DSL evaluation, use dg launch against a local project, or discover keys first with dg api asset list.
• --partition — single partition key. Partition ranges/backfills are not yet supported.
• --tag — tag to attach to the run as key=value. Repeatable.
• --config-json — JSON string of run config to use for the run
• --wait / -w — block until the run reaches a terminal status. Exits non-zero on FAILURE or CANCELED.
• --interval / -i — poll interval in seconds when --wait is set (default: 30)

At least one of --job or --asset-key must be provided.

dg api run list

• --status — filter by run status: QUEUED, STARTING, STARTED, SUCCESS, FAILURE, CANCELING, CANCELED. Repeatable (e.g. --status FAILURE --status CANCELED).
• --job — filter by job name

dg api schedule get-ticks <SCHEDULE_NAME>

• --status — filter by tick status: STARTED, SKIPPED, SUCCESS, FAILURE. Repeatable.
• --limit — maximum number of ticks to return (default: 25).
• --cursor — pagination cursor.
• --before — filter ticks before this Unix timestamp.
• --after — filter ticks after this Unix timestamp.
• --json — output in JSON format.

dg api schedule get <SCHEDULE_NAME>

Schedule API Commands

Commands for interacting with Dagster schedules via dg api schedule.

<!-- BEGIN GENERATED INDEX -->

• dg api schedule get-ticks — viewing schedule tick history; checking schedule evaluation results and failures
• dg api schedule get — details about a specific schedule
• dg api schedule list — listing schedules in Dagster Plus

<!-- END GENERATED INDEX -->

dg api schedule list

--status — filter schedules by status (e.g. RUNNING, STOPPED).

dg api secret get <SECRET_NAME>

• --show-value — includes the secret value in the response
• --location — filter by code location

dg api secret Reference

Commands for listing and inspecting secrets in a Dagster Plus deployment.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg api secret get — details about a specific secret
• dg api secret list — listing secrets in Dagster Plus

<!-- END GENERATED INDEX -->

dg api secret list

• --location — filter by code location
• --scope — filter by secret scope

dg api sensor get-ticks <SENSOR_NAME>

• --status — filter by tick status: STARTED, SKIPPED, SUCCESS, FAILURE. Repeatable.
• --limit — maximum number of ticks to return (default: 25).
• --cursor — pagination cursor.
• --before — filter ticks before this Unix timestamp.
• --after — filter ticks after this Unix timestamp.
• --json — output in JSON format.

dg api sensor get <SENSOR_NAME>

Sensor API Commands

Commands for interacting with Dagster sensors via dg api sensor.

<!-- BEGIN GENERATED INDEX -->

• dg api sensor get-ticks — viewing sensor tick history; checking sensor evaluation results and failures
• dg api sensor get — details about a specific sensor
• dg api sensor list — listing sensors in Dagster Plus

<!-- END GENERATED INDEX -->

dg api sensor list

--status — filter sensors by status (e.g. RUNNING, STOPPED).

dg check defs

Verify all definitions load without errors.

dg check defs
dg check defs --verbose    # Detailed output

dg check yaml

Validate defs.yaml files for syntax errors and valid component configuration.

dg check yaml

dg check toml

Validate pyproject.toml and dg.toml for syntax errors.

dg check toml

The create-dagster command scaffolds a new Dagster project with the proper Python package structure and Dagster-specific configuration.

Two structures are available:

• project — a single Dagster project (default choice unless user needs multiple independent packages)
• workspace — a collection of related Dagster projects with independent dependencies

IMPORTANT NEVER create a new Dagster project manually / without using the create-dagster command, as it will almost certainly be configured or structured improperly.

Project Creation

uvx create-dagster project <name> --uv-sync  # --uv-sync creates venv and installs deps (recommended)

Workspace Creation

uvx create-dagster workspace <name>          # For multiple related projects

Start a local Dagster development instance. This launches the Dagster webserver and daemon for local development.

dg dev

dg launch executes runs of assets or jobs locally and in-process. Useful for development. To launch a run on a remote Dagster Plus deployment, use `dg api run launch` instead.

dg launch --assets <selection>
dg launch --job <job_name>

See Asset Selection Syntax for complete selection syntax.

Partitions

dg launch --assets my_asset --partition 2024-01-15
dg launch --assets my_asset --partition-range "2024-01-01...2024-01-31"

Note: Use three dots (...) for inclusive ranges, not two dots.

Configuration

# Inline JSON
dg launch --assets my_asset --config '{"limit": 100}'

# From file
dg launch --assets my_asset --config-file config.yaml

List all available Dagster component types in the current Python environment.

dg list components

List all registered Dagster definitions (assets, jobs, schedules, sensors, resources) in the current project.

dg list defs

• --assets <selection> — filter by asset selection syntax
• --columns <cols> — columns to display (comma-separated or repeated flag)
• --json — output as JSON instead of a table
• --response-schema — print the JSON schema of the response and exit. Use before writing any parsing logic.

Available columns: key, group, deps, kinds, description, tags, cron, is_executable

dg list component-tree

List environment variables from the .env file of the current project. Shows variable name, whether it is set locally, and which components use it.

dg list envs

With Dagster Plus authentication (dg plus login), also shows deployment scope status (Dev/Branch/Full).

dg list Reference

Commands for exploring the structure of your Dagster project and workspace.

For listing definitions or available component types, see dg list defs and dg list components (top-level references).

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg list component-tree — viewing the component instance hierarchy
• dg list envs — seeing which environment variables the project requires
• dg list projects — listing projects in the current workspace

<!-- END GENERATED INDEX -->

dg list projects

Set configuration values for the Dagster Plus CLI.

dg plus config set <KEY> <VALUE>

Supported keys:

• api-token — set the API token for authentication
• org — set the organization name
• deployment — set the default deployment name
• url — set the Dagster Plus URL
• region — set the region (us or eu)

View the current Dagster Plus CLI configuration, including organization, deployment, API token, and URL.

dg plus config view

Create a CI/CD API token for automation workflows.

dg plus create ci-api-token

The generated token can be used in CI/CD pipelines to authenticate with Dagster Plus.

The dg plus deploy configure command will scaffold all the necessary files to allow a git repo to be deployed to Dagster Plus. At minimum, this will create a Github Actions or GitLab CI configuration file, which will automatically handle redeploying to Dagster Plus when commits are merged into the main branch / creating branch deployments for pull requests.

While the command does provide flags / subcommands for specific use cases, invoking it bare is _HIGHLY_ recommended, as this will prompt you for all necessary information, including authenticating with the relevant Git provider (GitHub or GitLab), as well as configuring container registry credentials if necessary (hybrid deployments).

dg plus login must ALWAYS be executed before running this command, which requires that the user create a Dagster Plus account.

IMPORTANT In order to successfully deploy to Dagster Plus, the dagster-cloud python package must be added as a dependency to the project (e.g. uv add dagster-cloud).

dg plus deploy configure

Deploy code to Dagster Plus. This command is typically invoked automatically by CI pipelines created via dg plus deploy configure, but can be run manually for ad-hoc deployments.

dg plus deploy

In most cases, prefer using dg plus deploy configure to set up automated CI/CD deployments rather than running this command directly.

Dagster Plus CLI Reference

Commands for authenticating with Dagster Plus, managing configuration, and deploying code.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg plus login — authenticating with Dagster Plus
• dg plus config set — set Dagster Plus CLI configuration values
• dg plus config view — view current Dagster Plus CLI configuration
• dg plus create ci-api-token — creating a CI/CD API token for Dagster Plus
• dg plus deploy configure — Deploying to Dagster Plus, Github Actions, GitLab CI; CI/CD configuration
• dg plus deploy — ad-hoc deployment to Dagster Plus
• dg plus integrations dbt — managing dbt manifests with Dagster Plus; downloading dbt manifest for dbt --defer or slim CI
• dg plus pull env — pulling environment variables from Dagster Plus into a local .env file

<!-- END GENERATED INDEX -->

Commands for managing dbt integrations with Dagster Plus.

dg plus integrations dbt manage-manifest

Automatically manage dbt manifest uploads and downloads based on deployment context. Used to enable dbt's --defer flag (slim CI) by providing a production state manifest. In branch deployments, downloads the prod manifest. In the source deployment (default: "prod"), uploads the manifest. Requires a dg plus deploy session (DAGSTER_BUILD_STATEDIR).

dg plus integrations dbt download-manifest

Download a dbt manifest from Dagster Plus for local development. Does not require a deploy session. Use --components to discover DbtProject instances from a dg project, or --file to point at a Python file containing DbtProject definitions. Use --output to override the download destination (cannot be used with multiple projects).

Authenticate with Dagster Plus. Opens a browser for interactive login.

dg plus login

--region eu — login to the European region (default is us).

Pull environment variables from Dagster Plus and save them to a local .env file. Requires authentication via dg plus login.

dg plus pull env

Scaffold a new custom Dagster component type class. Must be run inside a Dagster project directory. The scaffold is placed in <project_name>.components.<name>.

Use dg scaffold component when the component will be used multiple times. For one-off components, use dg scaffold defs inline-component instead, which places the component class definition directly under defs/ alongside its defs.yaml:

dg scaffold defs inline-component

dg scaffold component <class-name>

--model / --no-model — whether the generated class inherits from dagster.components.Model (default: --model).

Inspecting Components

To inspect an existing component type's description, scaffold parameters, or defs.yaml schema, use `dg utils inspect-component`.

dg scaffold defs is the preferred way to add new definitions to the project. It automatically ensures new code is added to the correct location.

Python Definition Objects

Scaffolds a single .py file at the specified path (relative to defs/). ALWAYS include the .py extension.

dg scaffold defs dagster.asset assets/my_asset.py
dg scaffold defs dagster.schedule schedules/daily.py
dg scaffold defs dagster.sensor sensors/watcher.py

Component Types

Scaffold a component directory with defs.yaml. Additional arguments can be provided via flags or --json-params.

Important: After scaffolding a custom component with dg scaffold component, run dg list components to get the exact registered type path. The path includes the file module name — e.g. my_project.components.my_component.MyComponent, not my_project.components.MyComponent.

dg scaffold defs some_lib.SomeComponent my_component

# With flags
dg scaffold defs dagster_dbt.DbtProjectComponent my_dbt --project-dir dbt_project

# With JSON params
dg scaffold defs dagster_dbt.DbtProjectComponent my_dbt --json-params '{"project_dir": "dbt_project"}'

Inline Components

For one-off components, use inline-component to place the component class definition directly under defs/ alongside its defs.yaml. Use dg scaffold component instead if you expect to reuse it.

dg scaffold defs inline-component

Important: Always run dg list defs to confirm the definitions were scaffolded correctly

Inspecting Components Before Scaffolding

To inspect a component's scaffold parameters or defs.yaml schema before scaffolding, use `dg utils inspect-component`.

dg utilities Reference

Utility commands for inspecting component types, and refreshing state-backed component cache.

Reference Files Index

<!-- BEGIN GENERATED INDEX -->

• dg utils inspect-component — inspecting a component type's description, schema, or examples
• dg utils refresh-defs-state — refreshing cached state for state-backed components

<!-- END GENERATED INDEX -->

dg utils inspect-component

Get detailed information about a registered component type, including its description, scaffold parameters, and configuration schema.

Usage

dg utils inspect-component <COMPONENT_TYPE>

Flags

All flags are mutually exclusive — only one can be used at a time.

• `--description` — Print the component's description text only.
• `--scaffold-params-schema` — Print the JSON schema for scaffold parameters (i.e. what flags/params dg scaffold defs accepts for this component type).
• `--defs-yaml-json-schema` — Print the full JSON schema for the component's defs.yaml file. This covers type, attributes, template_vars_module, requirements, and post_processing.
• `--defs-yaml-schema` — Print an LLM-optimized YAML template with inline documentation and type hints. Useful for understanding the structure at a glance.
• `--defs-yaml-example-values` — Print a YAML template populated with example values, useful for code generation.

With no flags, all available metadata is printed (description + scaffold params schema + component schema).

Refresh cached state for state-backed components. This fetches the latest state from external sources and updates the local cache.

dg utils refresh-defs-state

Creating Custom Components

Components are the primary unit of reuse in Dagster projects. A component is a Python class that maps YAML configuration to Dagster definitions via the Resolved framework. The core method is build_defs(), which returns a dg.Definitions object.

Scaffolding

Use the CLI to generate boilerplate for a new component:

dg scaffold component MyComponent

This creates the class file and registers it. Verify it appears in the component list, and note its full path (e.g. my_project.components.my_component.MyComponent) for future scaffolding:

dg list components

Component Structure

A component inherits from dg.Component and dg.Resolvable, plus a base class for field definitions.

See Resolved Framework for details on how to structure your component fields.

ALWAYS use the built-in resolved types for asset-related fields instead of raw strings or dicts:

• `dg.ResolvedAssetKey` — for a single asset key (accepts "a/b/c" string in YAML)
• `dg.ResolvedAssetSpec` — for a full asset spec (accepts structured mapping in YAML)
• `dg.ResolvedAssetCheckSpec` — for asset check specs

These handle YAML-to-Python resolution automatically.

Building Definitions

build_defs() returns dg.Definitions — this is the primary concern of a component.

Prefer @dg.multi_asset(specs=[...]) even for a single asset. This lets you pass AssetSpec objects directly via specs= instead of mapping all spec subfields to individual @dg.asset() kwargs:

import dagster as dg


class MyComponent(dg.Component, dg.Resolvable, dg.Model):
    spec: dg.ResolvedAssetSpec
    query: str

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        spec = self.spec

        @dg.multi_asset(specs=[spec])
        def my_asset(context: dg.AssetExecutionContext):
            context.log.info(f"Running query: {self.query}")
            # ... materialize the asset ...

        return dg.Definitions(assets=[my_asset])

Corresponding YAML:

type: my_project.components.MyComponent

attributes:
  spec:
    key: my_database/my_schema/orders
    group_name: ingestion
    kinds:
      - sql
  query: "SELECT * FROM orders"

Subsettable Multi-Assets

When a component produces multiple assets and the underlying tool supports executing an arbitrary subset independently, add can_subset=True to @dg.multi_asset() and mark each AssetSpec with skippable=True. Use context.selected_asset_keys to determine which assets to execute.

See Designing Component Integrations for the full pattern, including when to use subsetting vs. atomic execution.

Expensive Operations

If building definitions requires expensive work — querying a database, hitting an API, cloning a repo, compiling artifacts — ALWAYS use StateBackedComponent. It separates state-fetching from definition-building so that code server loads remain efficient.

If the external system already has a Dagster integration, prefer subclassing the existing component over building from scratch.

# Use StateBackedComponent instead of Component when external state is involved
class MyApiComponent(dg.StateBackedComponent, dg.Model, dg.Resolvable):
    ...

See State-Backed Components for full implementation details.

References

• Resolved Framework
• Template Variables
• State-Backed Components
• `dg scaffold component`
• `dg list components`

Designing Component Integrations

Before You Build: Check for Existing Integrations

Before designing a new component for an external tool, always check whether a built-in or community integration already exists in the Integrations index. Dagster ships integrations for 50+ tools including Fivetran, dbt, Snowflake, Power BI, Airbyte, Looker, Census, and more.

• If an integration exists: Subclass it and override only the methods you need to customize (get_asset_spec(), execute(), write_state_to_path()). See Subclassing Components.
• If no integration exists: You should always create a custom component rather than writing raw @dg.asset or @dg.sensor definitions. See Integration Workflow for the scaffolding steps, then continue with the patterns below.

Building a new component from scratch when an existing integration covers the same external system duplicates tested logic (API clients, state serialization, error handling) and misses future upstream improvements.

Three Levels of Integration

When designing a component integration, first determine how Dagster should interact with the external tool. There are three levels:

• Definition-only: Dagster understands assets defined in an external tool and their dependencies. Example: OmniComponent maps Omni dashboards to Dagster assets.
• Observing: Definition-only plus Dagster monitors for events via sensors, emitting AssetObservation or AssetMaterialization events. (Aspirational — no clean component example yet.)
• Orchestrating: Definition-only plus Dagster can trigger execution. Example: FivetranAccountComponent kicks off Fivetran syncs.

If it is necessary to fetch data from APIs in order to understand the _definitions_ of the assets, then State-Backed Components should always be used. If creating the definitions does NOT require fetching data (e.g. tool configuration is checked into the git repository), then a regular Component should be used, regardless of if executing the asset requires external API calls or not.

Pattern: External Data Class

Create a data class representing the raw data (for a specific asset) from the external tool's API. This is the "props" type that flows through translation and into get_asset_spec.

from dagster_shared.record import record

@record
class MyConnectorTableProps:
    """Raw data from the external tool for a single asset."""
    connector_id: str
    table_name: str
    schema_name: str
    sync_enabled: bool

Pattern: Translation Field

Translation allows YAML users to customize asset properties without subclassing. Use TranslationFnResolver with an Annotated type:

```python nocheck from typing import Annotated from dagster.components.utils.translation import TranslationFn, TranslationFnResolver

class MyServiceComponent(dg.Component, dg.Model, dg.Resolvable): translation: ( Annotated[ TranslationFn[MyConnectorTableProps], TranslationFnResolver( template_vars_for_translation_fn=lambda data: { "table_name": data.table_name, "schema_name": data.schema_name, } ), ] | None ) = None

`TranslationFn` is a type alias for `Callable[[AssetSpec, T], AssetSpec]`. The `template_vars_for_translation_fn` callback exposes fields as Jinja template variables for YAML users. The variable `spec` is always available automatically.

Corresponding YAML usage:

component_type: my_service params: translation: key: "my_prefix/{{ schema_name }}/{{ table_name }}" group: "{{ schema_name }}"

## Pattern: `get_asset_spec` Method

A public method that converts external data into a Dagster `AssetSpec`. It should provide sensible defaults (name → key, extract tags, set `kinds`, add metadata) and be designed for subclass override.

import dagster as dg

class MyServiceComponent(dg.Component, dg.Resolvable, dg.Model): translation: ...

def get_asset_spec(self, data: MyConnectorTableProps) -> dg.AssetSpec: """Generates an AssetSpec for a given connector table.""" base_spec = dg.AssetSpec( key=dg.AssetKey([data.schema_name, data.table_name]), metadata={"connector_id": data.connector_id}, kinds={"myservice"}, ) if self.translation: return self.translation(base_spec, data) return base_spec

Subclasses can override to customize defaults:

class CustomComponent(MyServiceComponent): def get_asset_spec(self, data: MyConnectorTableProps) -> dg.AssetSpec: spec = super().get_asset_spec(data) return spec.replace_attributes(group_name="my_group")

## Pattern: Credentials and Secrets

Component fields should accept credential **values** directly (e.g. `api_key: str`), not environment variable names (e.g. `api_key_env_var: str`). YAML users provide secrets via the Jinja `{{ env.VAR }}` syntax, which is resolved at component load time. The component code then uses the resolved value directly.

class MyServiceComponent(dg.Component, dg.Model, dg.Resolvable): api_secret: str

type: my_project.components.my_service_component.MyServiceComponent

attributes: api_secret: "{{ env.MY_SERVICE_API_SECRET }}"

See [Template Variables](./template-variables.md) for more information.

## Pattern: `execute()` Method

A public method for triggering external tool execution. Designed for subclass override.

def execute( self, context: dg.AssetExecutionContext, resource: MyServiceWorkspace ) -> Iterable[dg.AssetMaterialization | dg.MaterializeResult]: """Executes a sync for the selected connector.""" yield from resource.sync_and_poll(context=context)

The component wires `execute` into a `@multi_asset`:

def _build_multi_asset(self, connector_id, asset_specs, resource): @dg.multi_asset(name=connector_id, specs=asset_specs) def _assets(context: dg.AssetExecutionContext): yield from self.execute(context=context, resource=resource)

return _assets

## Pattern: Subsettable Multi-Assets

When the external tool supports executing an arbitrary subset of the assets defined in a single component instance, add `can_subset=True` to `@dg.multi_asset()` and use `context.selected_asset_keys` to determine which assets to execute.

**When to use**: The external tool lets you select which assets to execute independently (e.g. dbt lets you select any subset of models to build).

**When NOT to use**: The external tool executes all assets atomically (e.g. a Fivetran connector sync runs all tables together — no per-table control).

**Key changes**:

1. Add `can_subset=True` to the `@dg.multi_asset()` decorator in the `_build_multi_asset` helper
2. Mark each `AssetSpec` with `skippable=True` so Dagster knows individual assets can be skipped when subsetting
3. Pass `context` through to `execute()`:

def _build_multi_asset(self, connector_id, asset_specs, resource): @dg.multi_asset(name=connector_id, specs=asset_specs, can_subset=True) def _assets(context: dg.AssetExecutionContext): yield from self.execute(context=context, resource=resource)

return _assets

The `execute()` method can then use `context.selected_asset_keys` to only process the requested subset:

def execute( self, context: dg.AssetExecutionContext, resource: MyServiceWorkspace ) -> Iterable[dg.MaterializeResult]: for key in context.selected_asset_keys: yield from resource.sync_asset(key, context=context)

## Pattern: Observation via Sensors

For components that need to monitor external tool events without orchestrating them, a component's `build_defs` should include a sensor definition (adding this sensor could be controlled via a boolean flag on the component).

**Important**: Sensors for observing an external tool MUST be bundled inside the component's `build_defs()` method — do NOT create separate standalone `@dg.sensor` definition files for the same tool. The component should be the single source of truth for all definitions related to an integration, including sensors. This keeps observe vs. orchestrate mode toggling in one place (the component's YAML config).

def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions: asset_specs = [self.get_asset_spec(d) for d in []]

@dg.sensor(name="my_service_sensor") def _sensor(context: dg.SensorEvaluationContext): new_events = self._poll_for_events(context) for event in new_events: context.instance.report_runless_asset_event( dg.AssetObservation(asset_key=self.get_asset_spec(event).key) )

return dg.Definitions(assets=asset_specs, sensors=[_sensor] if self.enable_sensor else None)

## Cross-Component Dependencies

Components in the same code location can reference each other's assets by key — Dagster resolves all `deps` references across all components at load time. A downstream component only needs to know the **asset key** of an upstream asset, not where or how it's defined.

In a Census reverse ETL component — depends on dbt-produced marts

dg.AssetSpec( key=dg.AssetKey(["census", "salesforce_sync"]), deps=[dg.AssetKey(["snowflake", "marts", "mart_customers"])], # produced by DbtProjectComponent )

The `DbtProjectComponent` (or `FivetranAccountComponent`, etc.) is solely responsible for producing its own asset specs. Downstream components just reference the keys via `deps=`. There is no need to pre-define or duplicate asset specs for the dependency graph to connect.

Resolved Framework

Overview

The Resolved framework lets you write Pythonic classes that auto-generate YAML schemas with Jinja2 templating support. When a component field needs a non-primitive type (like AssetKey, datetime, or a custom object), you use Annotated with a Resolver to define how raw YAML values become Python objects.

Key classes (all importable via import dagster as dg):

• Resolvable — mixin that enables YAML-to-Python resolution on any class
• Model — pydantic BaseModel with extra="forbid", recommended for component attributes
• Resolver — metadata annotation that defines how a field is resolved from YAML

Choosing a Base Class

Both dg.Model and @dataclass are fully supported. Either works well for most components.

Model (pydantic) extends pydantic's BaseModel with extra="forbid". Benefits:

• Catches typos in YAML early (unknown fields are rejected)
• Field() metadata (descriptions, examples, defaults) propagates into the generated YAML schema, improving IDE autocompletion
• More powerful subclassing capabilities (subclasses can add new required fields)

Dataclass (@dataclass) is a lighter-weight alternative. Use field(default_factory=...) for mutable defaults. Dataclasses don't support Field() metadata propagation and have the standard Python limitation where subclasses cannot add required fields after optional ones.

Plain class with annotated `__init__` is supported for highly specific use cases, but is not recommended for standard use. The framework inspects the __init__ signature to derive fields.

Nested Resolution

The Resolved framework supports nesed resolution of classes, making it possible to simultaneously maintain complex python-native classes alongside automatically-generated YAML schemas.

There are a few options for nesting resoltion, depending on the specific use case.

Nested Resolvable Classes

Any class — not just components — can inherit dg.Resolvable + dg.Model to get automatic YAML resolution including Jinja2 template support. This is the preferred approach for structured config objects like connection configs, database configs, etc.

import dagster as dg


class ConnectionConfig(dg.Model, dg.Resolvable):
    token: str
    hostname: str


class MyComponent(dg.Component, dg.Resolvable, dg.Model):
    connection: ConnectionConfig  # auto-resolved, templates supported
    assets: list[dg.ResolvedAssetSpec]

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions: ...

Corresponding YAML — Jinja2 templates work inside nested Resolvable objects:

type: my_project.components.my_component.MyComponent

attributes:
  connection:
    token: "{{ env.TOKEN }}"
    hostname: "api.example.com"
  assets:
    - key: my_data/api_results

Annotated + Resolver

If your value contains a type that is not possible to natively represent in YAML, you can define a custom Resolver to handle this translation, and then use Annotated to create a new type alias that associates the custom resolver with the target python type.

from typing import Annotated, TypeAlias
from datetime import datetime
import dagster as dg


def resolve_datetime(context: dg.ResolutionContext, raw: str) -> datetime:
    resolved = context.resolve_value(raw, as_type=str)  # process templates first
    return datetime.fromisoformat(resolved)

ResolvedDatetime: TypeAlias = Annotated[datetime, dg.Resolver(resolve_datetime, model_field_type=str)]

class MyComponent(dg.Component, dg.Resolvable, dg.Model):
    # In YAML this field accepts a string; at load time it becomes a datetime
    start_date: ResolvedDatetime

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions: ...

*Args Classes

Sometimes, the target python type is a class that you cannot change the base class of (e.g. a third-party library class), but you still want to be able to resolve it from YAML.

In these cases, you can create a separate class that inherits from dg.Resolvable + dg.Model and mirrors the target python class's constructor signature.

from typing import Annotated, TypeAlias

# ... defined elsewhere ...
class SomeLibraryClass:

    def __init__(self, name: str, age: int): ...

class SomeLibraryClassArgs(dg.Resolvable, dg.Model):
    name: str
    age: int

def resolve_some_library_class(context: dg.ResolutionContext, model) -> SomeLibraryClass:
    # `model` will be an instance of SomeLibraryClassArgs.model()
    # this step will ensure all jinja templates are resolved (and handle any nested resolution)
    args = SomeLibraryClassArgs.resolve_from_model(context, model)
    # once the arguments are fully resolved, we can instantiate the target class using the resolved arguments
    return SomeLibraryClass(**args.model_dump())

ResolvedSomeLibraryClass: TypeAlias = Annotated[SomeLibraryClass, dg.Resolver(resolve_some_library_class, model_field_type=SomeLibraryClassArgs.model())]

class MyComponent(dg.Component, dg.Resolvable, dg.Model):
    some_library_class: ResolvedSomeLibraryClass

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions: ...

Resolver Types

`Resolver(fn)` — custom resolution function. The function receives (context: ResolutionContext, raw_value) and returns the resolved Python object. String values are template-resolved before the function is called (controlled by inject_before_resolve, default True).

`Resolver.default()` — standard recursive resolution. Resolves templates in the value and any nested Resolvable objects. Use this when the default behavior is sufficient but you want to add description or examples metadata:

from typing import Annotated

name: Annotated[
    str | None,
    dg.Resolver.default(
        description="Human-readable name of the asset.",
        examples=["my_asset"],
    ),
] = None

`Resolver.passthrough()` — returns the raw value without template processing or nested resolution. Use for fields that should receive the literal YAML value. This can be useful in cases where you intend to process jinja templates _after_ the component has been loaded.

`Resolver.from_model(fn)` — the function receives the entire parent model instead of just the field value. Use when resolution depends on multiple fields together.

Injected[T] is a shorthand for Annotated[T, Resolver.default(model_field_type=str)] — the field accepts a string in YAML (for template injection) and resolves to type T.

Built-in Type Aliases

`dg.ResolvedAssetKey` accepts a string like "my_database/my_schema/my_table" in YAML and resolves to an AssetKey. Template strings are resolved before parsing.

`dg.ResolvedAssetSpec` accepts a structured mapping in YAML (with fields like key, deps, group_name, tags, kinds, automation_condition, partitions_def) and resolves to an AssetSpec.

`dg.ResolvedAssetCheckSpec` accepts a structured mapping and resolves to an AssetCheckSpec.

How Schema Generation Works

When you define a Resolvable class, the framework auto-generates a pydantic model for YAML validation:

1. Each field's type and Resolver metadata are inspected 2. If model_field_type is set on the Resolver, that type is used in the schema instead of the Python type 3. All non-str fields become field_type | str in the schema, allowing any field to accept a Jinja2 template string 4. description and examples from the Resolver propagate into the generated schema, appearing in IDE autocompletion during YAML editing

References

• Template Variables