Create Custom Dagster Component

Create a custom Component

Overview

This skill automates the creation and validation of a new custom Dagster component using the dg CLI tool with uv as a package manager. It incorporates demo mode functionality for creating realistic demonstrations that can run locally without external dependencies. The documentation for creating good components can be found here https://docs.dagster.io/guides/build/components/creating-new-components/creating-and-registering-a-component and here https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-dbt/dagster_dbt/components/dbt_project/component.py for a complex example of a 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
sources:
  - name: api_raw
    tables:
      - name: customers  # Matches ["api_raw", "customers"]

Verification

Always verify asset keys align with downstream dependencies:

# Check asset keys and their dependencies
uv run dg list defs --json | uv run python -c "
import sys, json
assets = json.load(sys.stdin)['assets']
print('\\n'.join([f\"{a['key']}: deps={a.get('deps', [])}\" for a in assets]))
"

What to verify:

• Downstream assets list your assets in their deps array
• No duplicate keys with different structures
• Keys are simple and descriptive (typically 2 levels: ["category", "name"])

Anti-Patterns to Avoid

❌ Too deeply nested: ["company", "team", "project", "environment", "table"]

• Hard for downstream to reference
• Requires complex mapping

❌ Inconsistent structure: Some assets with 2 levels, others with 4

• Confusing for consumers
• Unpredictable references

❌ Generic names: ["data", "table1"], ["output", "result"]

• Not clear what system they're from
• Conflicts with other components

✅ Good patterns:

• ["source_system", "entity"]: ["fivetran_raw", "customers"]
• ["integration", "object"]: ["salesforce", "accounts"]
• ["stage", "table"]: ["staging", "orders"]

Critical: Asset Keys Must Be Identical in Demo and Production Mode

IMPORTANT: Asset keys should be exactly the same whether demo_mode is True or False. Only the asset implementation (the function body) should differ between modes.

Why this matters:

• Downstream components reference assets by key
• Dependencies are established based on keys
• If keys differ between modes, dependencies break when switching modes
• Testing in demo mode won't accurately reflect production behavior

Example - CORRECT approach:

def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
    @dg.asset(
        key=dg.AssetKey(["fivetran_raw", "customers"]),  # ← Same key in both modes
        kinds={"fivetran"},
    )
    def customers_sync(context: dg.AssetExecutionContext):
        if self.demo_mode:
            # Demo implementation - mock data
            context.log.info("Demo mode: Creating empty table")
            # ... create mock table
        else:
            # Production implementation - real Fivetran sync
            context.log.info("Production: Syncing from Fivetran")
            # ... call Fivetran API

    return dg.Definitions(assets=[customers_sync])

Example - INCORRECT approach:

def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
    if self.demo_mode:
        @dg.asset(
            key=dg.AssetKey(["demo", "customers"]),  # ❌ Different key!
        )
        def demo_customers():
            pass
        return dg.Definitions(assets=[demo_customers])
    else:
        @dg.asset(
            key=dg.AssetKey(["fivetran_raw", "customers"]),  # ❌ Different key!
        )
        def prod_customers():
            pass
        return dg.Definitions(assets=[prod_customers])

Reference Documentation:

• Cross-reference https://docs.dagster.io/llms.txt for up-to-date titles and descriptions
• Use https://docs.dagster.io/llms-full.txt for full API details
• Check available integrations with:

uv run dg docs integrations --json

Important: Always Add Kinds to Assets

When creating assets in your component, ALWAYS add the kinds parameter to properly categorize assets by their technology/integration type. This helps with:

• Filtering and organizing assets in the Dagster UI
• Understanding the technology stack at a glance
• Grouping assets by integration type

Common integration kinds:

• kinds={"fivetran"} for Fivetran assets
• kinds={"dbt"} for dbt assets
• kinds={"census"} for Census assets
• kinds={"sling"} for Sling assets
• kinds={"powerbi"} for PowerBI assets
• kinds={"looker"} for Looker assets
• kinds={"airbyte"} for Airbyte assets
• kinds={"python"} for custom Python processing
• kinds={"snowflake"} for Snowflake assets

You can verify kinds are showing correctly by running:

uv run dg list defs

The "Kinds" column should show the integration type for each asset.

Example Component Structure:

from dagster import asset, Definitions, AssetExecutionContext
from pydantic import BaseModel

class MyComponentParams(BaseModel):
    demo_mode: bool = False
    # ... other params

class MyComponent(Component):
    params_schema = MyComponentParams

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

        @asset(
            kinds={"fivetran"},  # ← REQUIRED: Add the integration kind
        )
        def raw_data(context: AssetExecutionContext):
            if params.demo_mode:
                # Demo implementation - local/mocked data
                context.log.info("Running in demo mode with local data")
                pass
            else:
                # Real implementation - connect to actual systems
                context.log.info("Running with real data source")
                pass

        @asset(
            deps=[raw_data],
            kinds={"dbt"},  # ← REQUIRED: Add the integration kind
        )
        def processed_data(context: AssetExecutionContext):
            if params.demo_mode:
                context.log.info("Processing demo data")
                pass
            else:
                context.log.info("Processing real data")
                pass

        # ... more assets

        return Definitions(assets=[raw_data, processed_data, ...])

Step 4: Create Component Instance YAML

Use dg scaffold defs to create the component instance:

uv run dg scaffold defs my_module.components.ComponentName my_component

This creates a YAML file that should include the demo_mode parameter:

type: my_module.components.ComponentName
attributes:
  demo_mode: true  # Set to true for local demos, false for real deployments
  # ... other params

Step 5: Create Custom Scaffolder (Optional)

If the user requested a custom scaffolder in Step 1, follow the directions here: https://docs.dagster.io/guides/build/components/creating-new-components/component-customization#customizing-scaffolding-behavior

Customize the scaffolder to provide a better developer experience for creating instances of this component.

Step 6: Validate Setup and Asset Key Alignment

Run these commands to ensure everything works:

# Check that definitions load without errors
uv run dg check defs

# List all assets to verify they were created
uv run dg list defs

Verify that:

• ✅ All expected assets are listed
• ✅ The component instance is properly configured
• ✅ No errors or warnings are shown
• ✅ The demo_mode flag toggles between implementations correctly
• ✅ The demo_mode: false implementation uses realistic resources and is a production implementation

CRITICAL: Verify Asset Key Alignment

Check that asset dependencies are correct by running:

uv run dg list defs --json | uv run python -c "
import sys, json
data = json.load(sys.stdin)
assets = data.get('assets', [])
print('Asset Dependencies:\n')
for asset in assets:
    key = asset.get('key', 'unknown')
    deps = asset.get('deps', [])
    if deps:
        print(f'{key}')
        for dep in deps:
            print(f'  ← {dep}')
    else:
        print(f'{key} (no dependencies)')
    print()
"

What to verify:

• ✅ Downstream assets list upstream assets in their deps array
• ✅ No missing dependencies
• ✅ Asset keys are simple and descriptive (typically 2 levels: ["category", "name"])
• ✅ Asset keys work consistently in both demo mode and production mode

Key Principle: Asset keys should be identical between demo mode and production mode. Only the asset implementation (the function body) should differ. This ensures:

• Dependencies work the same in both modes
• You can switch between modes without reconfiguring downstream components
• Testing in demo mode accurately reflects production behavior

Step 7: Test Demo Mode

If demo mode was implemented:

1. Ensure the component YAML has demo_mode: true
2. Run dg check defs to verify it works locally
3. Document how to switch between demo and real modes

Success Criteria

The component is complete when:

• ✅ Component scaffolding is created
• ✅ build_defs() is implemented with proper asset logic
• ✅ Demo mode flag is working (if applicable)
• ✅ Non demo mode has realistic connections to the database or APIs implemented
• ✅ 3-5 realistic assets are created with proper dependencies
• ✅ Assets have appropriate kinds metadata
• ✅ Component YAML instance is created and configured
• ✅ Custom scaffolder is implemented (if requested)
• ✅ dg check defs passes without errors
• ✅ dg list defs shows all expected assets

Next Steps

After completion, inform the user:

1. The component has been created and validated
2. Location of the component files
3. How to toggle demo mode (if applicable)
4. How to customize the component further
5. How to create additional instances using the scaffolder