Use Or Subclass Existing Component

When invoked, this skill will:

1. ✅ Help discover available Dagster integrations using uv run dg docs integrations --json or browsing https://docs.dagster.io/integrations/libraries
2. ✅ Determine whether to use the component directly or create a subclass
3. ✅ If subclassing: Create a local component subclass that extends the integration component
4. ✅ Implement demo mode support by overriding the execute() method
5. ✅ Add custom templating scope if needed via get_additional_scope()
6. ✅ Create and configure component instance YAML with proper type references
7. ✅ Validate that the component loads correctly using uv run dg check defs
8. ✅ Verify assets are created using uv run dg list defs

Prerequisites

Before running this skill, ensure:

• uv is installed (check with uv --version)
• You're in a Dagster project directory with the dg CLI available
• You know which integration you want to use (e.g., dbt, Looker, PowerBI, Fivetran)

Important: When to Switch Skills

If you discover the integration package exists but has NO Component class:

• Run: uv run python -c "import dagster_<integration>; print([x for x in dir(dagster_<integration>) if 'Component' in x])"
• If result is [], the integration doesn't have a Component class
• STOP and use the create-custom-dagster-component skill instead

This skill is ONLY for integrations that have existing Component classes to subclass.

Skill Workflow

Step 1: Discover Available Integrations

Help the user find the right integration component:

1. Run discovery command:

   uv run dg docs integrations --json
   

2. Alternative: Browse https://docs.dagster.io/integrations/libraries for visual list

3. Common integrations include:

   • dagster_dbt.DbtProjectComponent - dbt projects
   • dagster_fivetran.FivetranComponent - Fivetran syncs
   • dagster_sling.SlingReplicationCollectionComponent - Sling replications
   • dagster_powerbi.PowerBIWorkspaceComponent - PowerBI workspaces
   • dagster_looker.LookerComponent - Looker instances
   • dagster_airbyte.AirbyteComponent - Airbyte connections
   • dagster_databricks.DatabricksComponent - Databricks workflows
   • dagster_snowflake.SnowflakeComponent - Snowflake resources

Step 2: Determine Component Type and Use Case

IMPORTANT: First determine if this is a configuration-file-based or API-based component:

Configuration-File-Based Components

These components read from local configuration files and do NOT require external API credentials:

• DbtProjectComponent - Reads from dbt project files (dbt_project.yml, SQL/Python models)
• SlingReplicationCollectionComponent - Reads from replication YAML files
• Components that primarily process local files

For configuration-file-based components:

• ✅ DO use the component directly or subclass for custom behavior
• ❌ DO NOT implement demo_mode - just load the project files
• The component naturally works with local files without external dependencies

API-Based Components

These components call out to external services and require API credentials:

• FivetranComponent - Calls Fivetran API
• PowerBIWorkspaceComponent - Calls PowerBI API
• LookerComponent - Calls Looker API
• AirbyteComponent - Calls Airbyte API
• CensusComponent - Calls Census API

For API-based components:

• ✅ DO implement demo_mode to mock external API calls
• ✅ DO provide dummy credentials in YAML (ignored when demo_mode is true)
• This allows the component to work locally without external connections

Determine Your Use Case

Based on the component type, choose:

1. Use directly (Configuration-based) - Load the component with local configuration files
2. Use directly (API-based) - Use the component with real API credentials
3. Subclass for demo mode (API-based only) - Add demo_mode to mock external API calls
4. Subclass for other customization - Add other custom behavior beyond demo mode

If using directly, skip to Step 7 to create the component instance YAML.

Step 3: Install Integration Package

Install the required Dagster integration package:

uv add dagster-<integration-name>

Examples:

• uv add dagster-dbt
• uv add dagster-sling
• uv add dagster-powerbi

Step 4: Scaffold Component Instance

Use dg scaffold defs to create the component instance directory:

uv run dg scaffold defs <package>.<ComponentClass> <instance_name>

Example:

uv run dg scaffold defs dagster_sling.SlingReplicationCollectionComponent my_sling_sync

This creates a directory structure like:

defs/
  my_sling_sync/
    defs.yaml
    # Other config files as needed

Step 5: Create Component Subclass

Create a component.py file in the component instance directory.

IMPORTANT: Only add demo_mode for API-based components!

• Configuration-file-based components (dbt, Sling) read from local files and don't need demo_mode
• API-based components (Fivetran, PowerBI, Looker) call external APIs and should use demo_mode

When extending integration components, you must understand whether they use dataclass or Pydantic BaseModel patterns, as this determines how to add custom fields like demo_mode.

Identifying Component Type

First, check the parent component's implementation:

# Check if it's a dataclass
uv run python -c "from <package> import <Component>; import dataclasses; print(dataclasses.is_dataclass(<Component>))"

Most Dagster integration components (like SlingReplicationCollectionComponent, DbtProjectComponent, FivetranComponent) use dataclass with the Resolvable interface.

Pattern for Dataclass-Based Components (Recommended)

This is the most common pattern for Dagster integration components:

from dataclasses import dataclass
import dagster as dg
from <integration_package> import <BaseComponentClass>

@dataclass
class Custom<ComponentName>(BaseComponentClass):
    """Customized component with demo mode support."""

    # New field - will automatically appear in YAML schema via Resolvable
    demo_mode: bool = False

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        """Build definitions, using demo mode if enabled.

        Note: The parent class fields (like API credentials) are still set from YAML,
        but when demo_mode is True, we bypass the parent's build_defs() method
        and return mocked assets instead, so those credentials are never used.
        """
        if self.demo_mode:
            # Return mock assets for demo mode - parent credentials are ignored
            return self._build_demo_defs(context)
        else:
            # Use real integration with actual credentials from parent fields
            return super().build_defs(context)

    def _build_demo_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        """Build demo mode definitions with mocked assets."""
        @dg.asset(
            key=dg.AssetKey(["mock_asset"]),
            kinds={"integration_name"},  # IMPORTANT: Add the integration kind
        )
        def mock_asset(context: dg.AssetExecutionContext):
            context.log.info("Demo mode: simulating asset execution")
            return {"status": "demo_mode"}

        return dg.Definitions(assets=[mock_asset])

Key points for dataclass components:

1. Use @dataclass decorator on your subclass
2. Add fields with type annotations - they automatically become YAML schema fields
3. Use field(default_factory=...) for mutable defaults (lists, dicts)
4. The Resolvable interface (inherited from parent) handles YAML schema generation
5. All fields with defaults must come after fields without defaults

Example with multiple custom fields:

from dataclasses import dataclass, field
import dagster as dg
from dagster_sling import SlingReplicationCollectionComponent

@dataclass
class CustomSlingComponent(SlingReplicationCollectionComponent):
    """Extended Sling component with additional configuration."""

    # New fields - all will appear in YAML schema
    demo_mode: bool = False
    enable_notifications: bool = False
    notification_channel: str = "slack"
    custom_tags: list[str] = field(default_factory=list)

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        if self.demo_mode:
            return self._build_demo_defs(context)
        return super().build_defs(context)

Pattern for Pydantic BaseModel Components (Rare)

Some components may use Pydantic BaseModel. In these cases, inherit from both the parent component and dg.Model:

import dagster as dg
from <integration_package> import <BaseComponentClass>

class Custom<ComponentName>(BaseComponentClass, dg.Model):
    """Customized component with demo mode support."""

    # New field - will appear in YAML schema
    demo_mode: bool = False

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        if self.demo_mode:
            return self._build_demo_defs(context)
        return super().build_defs(context)

Pattern for Legacy execute()-based Components

For older components that override execute() instead of build_defs():

from dataclasses import dataclass
from dagster import AssetExecutionContext
from <integration_package> import <BaseComponentClass>
from collections.abc import Iterator
from typing import Any

@dataclass
class Custom<ComponentName>(BaseComponentClass):
    """Customized component with demo mode support."""

    demo_mode: bool = False

    def execute(
        self,
        context: AssetExecutionContext,
        **kwargs: Any,
    ) -> Iterator:
        """Custom execution logic with demo mode support."""
        if self.demo_mode:
            context.log.info("Running in demo mode with mocked data")
            yield from self._execute_demo_mode(context, **kwargs)
        else:
            context.log.info("Running with real integration")
            yield from super().execute(context, **kwargs)

    def _execute_demo_mode(
        self,
        context: AssetExecutionContext,
        **kwargs: Any,
    ) -> Iterator:
        """Demo mode implementation."""
        from dagster import Output
        context.log.info("Simulating integration execution locally")
        yield Output(value=None, output_name="result")

Key customization points:

1. Override build_defs() or execute() - Check demo_mode and return mock data
2. Use dataclass fields - They automatically become YAML schema fields via Resolvable
3. Provide default values - All custom fields should have defaults
4. Add custom templating - Override get_additional_scope() for YAML templating

Example with custom templating:

from dataclasses import dataclass
from collections.abc import Mapping
from typing import Any
import dagster as dg
from <integration_package> import <BaseComponentClass>

@dataclass
class Custom<ComponentName>(BaseComponentClass):
    demo_mode: bool = False

    @classmethod
    def get_additional_scope(cls) -> Mapping[str, Any]:
        """Add custom YAML templating functions."""
        def _custom_cron(cron_schedule: str) -> dg.AutomationCondition:
            return (
                dg.AutomationCondition.on_cron(cron_schedule)
                & ~dg.AutomationCondition.in_progress()
            )
        return {"custom_cron": _custom_cron}

Step 5.5: Align Asset Keys for Downstream Components (CRITICAL for Multi-Component Pipelines)

When to do this: If other Dagster components in your pipeline will depend on assets from this component, override get_asset_spec() to generate asset keys that match downstream expectations.

This applies when:

• dbt models will consume these assets
• Custom Dagster assets will reference these assets in their deps
• Another integration component (Hightouch, Census, BI tools) expects specific asset key patterns
• You want to simplify multi-level nested keys for easier downstream consumption

Why This Matters

By default, integration components generate asset keys in their own structure. For example:

• Fivetran: ["fivetran", "raw", "customers"]
• Sling: ["sling", "replications", "sync_name", "table"]
• dbt: ["analytics", "marts", "customer_360"]

The problem: Downstream components may expect different key structures, leading to broken dependencies or requiring per-asset configuration with meta.dagster.asset_key.

The solution: Override get_asset_spec() in the upstream component to generate keys that downstream components naturally reference.

Pattern: Override get_asset_spec()

def get_asset_spec(self, props) -> dg.AssetSpec:
    """Override to generate asset keys matching downstream component expectations.

    This eliminates the need for meta.dagster.asset_key configuration in downstream
    components by aligning keys at the source.
    """
    base_spec = super().get_asset_spec(props)
    original_key = base_spec.key.path

    # Customize key structure for your pipeline
    # Example: Flatten nested keys for easier consumption
    custom_key = dg.AssetKey([...])  # Your key transformation logic

    return base_spec.replace_attributes(key=custom_key)

Example 1: Fivetran → dbt Pipeline

Problem: Fivetran creates ["fivetran", "raw", "customers"], but dbt expects ["fivetran_raw", "customers"]

Solution:

from dagster_fivetran import FivetranAccountComponent
from dagster_fivetran.translator import FivetranConnectorTableProps
import dagster as dg

class CustomFivetranComponent(FivetranAccountComponent):
    def get_asset_spec(self, props: FivetranConnectorTableProps) -> dg.AssetSpec:
        """Flatten asset keys for dbt compatibility."""
        base_spec = super().get_asset_spec(props)
        original_key = base_spec.key.path

        # Flatten: ["fivetran", "raw", "table"] -> ["fivetran_raw", "table"]
        if len(original_key) >= 2:
            flattened_key = dg.AssetKey(["fivetran_raw", "_".join(original_key[1:])])
        else:
            flattened_key = dg.AssetKey(["fivetran_raw", original_key[-1]])

        return base_spec.replace_attributes(key=flattened_key)

Result: dbt sources work automatically without meta.dagster configuration:

# sources.yml - references work naturally now