dagster

Project Structure

Start with a single definitions.py. Split when files exceed ~400 lines:

src/my_project/
├── __init__.py
├── definitions.py      # Definitions object — single entry point
├── assets.py           # @asset functions
├── resources.py        # ConfigurableResource classes
├── schedules.py
├── sensors.py
└── defs/               # Subdirectories by domain or technology
    ├── ingestion/
    └── reporting/

Use load_assets_from_package_module() to auto-discover assets from subdirectories.

Assets

from dagster import asset, AssetIn, AssetCheckResult, asset_check

@asset(group_name="ingestion", key_prefix="raw")
def raw_orders(db: DatabaseResource) -> pd.DataFrame:
    return db.query("SELECT * FROM orders")

@asset(ins={"raw_orders": AssetIn(key_prefix="raw")})
def cleaned_orders(raw_orders: pd.DataFrame) -> pd.DataFrame:
    return raw_orders.dropna(subset=["order_id"])

@asset_check(asset=cleaned_orders)
def no_null_ids(cleaned_orders: pd.DataFrame) -> AssetCheckResult:
    passed = cleaned_orders["order_id"].notna().all()
    return AssetCheckResult(passed=passed)

• Keep asset functions thin — delegate complex logic to utility modules.
• Use group_name and key_prefix for organization.
• Add asset_check for data quality validation.

Resources

from dagster import ConfigurableResource

class DatabaseResource(ConfigurableResource):
    connection_string: str

    def query(self, sql: str) -> pd.DataFrame:
        ...

• Extend ConfigurableResource for all external connections.
• Register resources in one Definitions object.
• Swap implementations per environment (production vs. test).

Sensors & Schedules

• Prefer declarative automation over asset sensors when only triggering asset execution.
• Test sensors with build_sensor_context and validate_run_config.
• Test schedules with build_schedule_context.

Partitions

• Time-based for temporal data, static for fixed categories, dynamic for runtime sets.
• Keep partition count under 100,000 per asset for UI performance.

Testing

def test_cleaned_orders():
    raw = pd.DataFrame({"order_id": [1, None, 3], "amount": [10, 20, 30]})
    result = cleaned_orders(raw)
    assert result["order_id"].notna().all()
    assert len(result) == 2

• Invoke asset functions directly with test inputs and mocked resources.
• Use build_asset_context() when assets need context.
• Test sensors/schedules: call the decorated function, validate with validate_run_config.

Anti-Patterns

| Avoid | Do Instead | |---|---| | Hardcoded connection strings in assets | Use ConfigurableResource | | Business logic inside @asset functions | Extract to utility modules, keep assets as orchestration glue | | materialize_to_memory in production | Use Definitions with proper resources | | Monolithic definitions.py beyond 400 lines | Split by domain or technology | | Ignoring asset checks | Add checks for key data quality invariants |

When to Use

Load this skill when building or modifying Dagster pipelines, assets, resources, sensors, or schedules.