Create Custom Dagster Component

What This Skill Does

When invoked, this skill will:

1. ✅ Create a new Dagster Component project using dg scaffold component ComponentName
2. ✅ Fills in the component logic in the build_defs() function with both real and demo mode implementations
3. ✅ Implement a demo_mode boolean flag in the component YAML for toggling between real and local demo implementations
4. ✅ Create 3-5 realistic assets with proper dependencies and technology kinds
5. ✅ Instantiate a component YAML and fill it in using the dg scaffold defs my_module.components.ComponentName my_component command
6. ✅ Optionally create a custom scaffolder if requested
7. ✅ Validate that the component loaded correctly using dg check defs and dg list defs to ensure that the expected component instances are all loaded.
8. ✅ Provide clear next steps for development

Prerequisites

Before running this skill, ensure:

• uv is installed (check with uv --version)
• You have a component name in mind (or will use the default)
• You're in a Dagster project directory with the dg CLI available
• You have inspected and understand how to create multi-asset integrations (see this guide: https://docs.dagster.io/integrations/guides/multi-asset-integration)

Skill Workflow

Step 1: Get Component Name and Demo Mode Preference

Ask the user for:

1. A component name, or use a sensible default like MyDagsterComponent. Validate that:
   • The name starts with a letter
   • Contains only alphanumeric characters, hyphens, or underscores
   • The component doesn't already exist (or ask to overwrite)
2. Whether they want demo mode support (default: yes for demonstration projects)
3. Whether they want to create a custom scaffolder (see Step 5)

Step 2: Create Component

Use dg to create the component

uv run dg scaffold component <ComponentName>

This will:

• Scaffold a new Dagster Component in defs/components
• Create a component_name.py file

Step 3: Implement Component Logic with Demo Mode

Fill in the build_defs() function in the component file. The component should:

1. Accept a demo_mode parameter in the component params (default: False)
2. Create 3-5 realistic assets based on the chosen technologies
3. Implement dual logic paths:
   • Real implementation: connects to actual systems (the body should include connections to real systems)
   • Demo mode: uses local data/mocked behavior for demonstrations
   • Important Configuration in Pydantic + YAML fields: All configuration (a new pipeline or asset) should be configurable in the Component and not hard coded in the Component Python. Demo mode assets are the exception that should be hard coded.
   • All Resources should be configured outside of the Component in the defs/ folder in a resources.py file, using dg scaffold defs dagster.resource resources.py
   • All Components that invoke a pipeline or API that might trigger multiple assets should allow for an assets field in the YAML that describes what assets are used in the underlying component. See https://dagster.io/blog/dsls-to-the-rescue for best practices in how to design a good DSL. Refer to https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-dbt/dagster_dbt/components/dbt_project/component.py and https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-fivetran/dagster_fivetran/components/workspace_component/component.py for two reference architectures for good component design with mutli-assets.
4. Set proper asset metadata:
   • Use the kinds argument to indicate technologies in use
   • Add descriptive names and documentation
   • Establish proper dependencies between assets
5. Design asset keys for downstream integration (CRITICAL):
   • Consider what components will consume your assets
   • Choose key structures that minimize downstream configuration
   • See "Design Asset Keys for Integration" section below

Example asset structure:

• Raw data ingestion asset
• Data transformation/cleaning asset
• Business logic/aggregation asset
• ML model or analytics asset (if applicable)
• Output/export asset

Design Asset Keys for Integration

CRITICAL: When creating a custom component, consider what will consume your component's assets. The asset keys you generate should align with downstream component expectations to avoid requiring per-asset configuration.

Key Principle: Upstream Defines, Downstream Consumes

Your component (upstream) should generate asset keys in a structure that downstream components naturally reference. This eliminates the need for meta.dagster.asset_key or complex translation configuration.

Common Downstream Consumers

If dbt will consume your assets:

• Use pattern: ["<source_name>", "<table_name>"]
• Example: ["fivetran_raw", "customers"] or ["api_raw", "users"]
• This allows dbt sources to reference naturally: source('fivetran_raw', 'customers')

If custom Dagster assets will consume them:

• Match the key structure those assets expect in their deps
• Minimize nesting when possible (prefer 2 levels: ["category", "name"])
• Avoid deeply nested keys like ["system", "subsystem", "type", "name"] unless necessary

If another integration component will consume them:

• Check that component's expected input key structure
• Align your keys to match, or provide clear mapping documentation

If your assets are intermediate and consumed by your own component:

• Use clear, hierarchical keys that reflect data flow
• Example: ["raw", "table"] → ["processed", "table"] → ["enriched", "table"]

Example: Creating an API Ingestion Component

import dagster as dg

class APIIngestionComponent(dg.Component, dg.Model, dg.Resolvable):
    """Ingests data from REST APIs."""

    api_endpoint: str
    tables: list[str]
    demo_mode: bool = False

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        assets = []

        for table in self.tables:
            # Design key for dbt consumption: ["api_raw", "table_name"]
            # NOT: ["api", "ingestion", "raw", "table_name"]
            @dg.asset(
                key=dg.AssetKey(["api_raw", table]),  # ← Flattened for easy downstream reference
                kinds={"api", "python"},
            )
            def ingest_table(context: dg.AssetExecutionContext):
                if self.demo_mode:
                    context.log.info(f"Demo mode: Mocking API call for {table}")
                    return {"status": "demo", "rows": 100}
                else:
                    # Real API call
                    pass

            assets.append(ingest_table)

        return dg.Definitions(assets=assets)

Result: dbt can reference these assets naturally:

# sources.yml