Skip to main content

Components

Using Components​

@dagster.component_instance [source]

Decorator for a function to be used to load an instance of a Component. This is used when instantiating components in python instead of via yaml.

Example:

import dagster as dg

class MyComponent(dg.Component):
    ...

@dg.component_instance
def load(context: dg.ComponentLoadContext) -> MyComponent:
    return MyComponent(...)
class dagster.ComponentLoadContext [source]

Context object that provides environment and path information during component loading. This context is automatically created and passed to component definitions when loading a project’s defs folder. Each Python module or folder in the defs directory receives a unique context instance that provides access to the underlying ComponentDecl, project structure, paths, and utilities for dynamic component instantiation.

The context enables components to:

  • Access project and module path information
  • Load other modules and definitions within the project
  • Resolve relative imports and module names
  • Access templating and resolution capabilities

Parameters:

  • path – The filesystem path of the component currently being loaded. For a file: /path/to/project/src/project/defs/my_component.py For a directory: /path/to/project/src/project/defs/my_component/
  • project_root – The root directory of the Dagster project, typically containing pyproject.toml or setup.py. Example: /path/to/project
  • defs_module_path – The filesystem path to the root defs folder. Example: /path/to/project/src/project/defs
  • defs_module_name – The Python module name for the root defs folder, used for import resolution. Typically follows the pattern "project_name.defs". Example: "my_project.defs"
  • resolution_context – The resolution context used by the component templating system for parameter resolution and variable substitution.
  • component_tree – The component tree that contains the component currently being loaded.
  • terminate_autoloading_on_keyword_files – Controls whether autoloading stops when encountering definitions.py or component.py files. Deprecated: This parameter will be removed after version 1.11.
  • component_decl – The associated ComponentDecl to the component being loaded.

Note: This context is automatically provided by Dagster’s autoloading system and should not be instantiated manually in most cases. For testing purposes, use ComponentTree.for_test().load_context to create a test instance.

See also: - dagster.definitions(): Decorator that receives this context

class dagster.ComponentTree [source]

The hierarchy of Component instances defined in the project.

Manages and caches the component loading process, including finding component declarations to build the initial declaration tree, loading these Components, and eventually building the Definitions.

Building Components​

@dagster.template_var [source]

Decorator that marks a function as a template variable for use in component YAML definitions.

Template variables provide dynamic values and functions that can be injected into component YAML definitions using Jinja2 templating syntax ({{ variable_name }}). They are evaluated at component load time and can optionally receive a ComponentLoadContext parameter for context-aware behavior.

These values can be any python object and are passed directly to the component as Python object. They can be injected at any level of the defs file.

There are two main usage patterns:

  1. Module-level template variables: Functions defined in a separate module and referenced via the template_vars_module field in component YAML
  2. Component class static methods: Template variables defined as @staticmethod on a Component class, automatically available to instances of that component Template vars can themselves be functions, in which case they are user-defined functions, invoked with function syntax within the defs file.

Parameters: fn – The function to decorate as a template variable. If None, returns a decorator.Returns: The decorated function with template variable metadata, or a decorator function. Note: Template variables are evaluated at component load time, not at runtime. They provide configuration values and functions for YAML templating, not runtime component logic.

Function Signatures: Template variable functions can have one of two valid signatures:

Zero parameters (static values):

@dg.template_var
def static_value() -> Any:
    # Returns a static value computed at load time
    return "computed_value"

Single ComponentLoadContext parameter (context-aware):

@dg.template_var
def context_value(context: dg.ComponentLoadContext) -> Any:
    # Returns a value based on the component's loading context
    return f"value_{context.path.name}"

Return Types: Template variables can return any type, including:

  • Primitive values: str, int, bool, float
  • Collections: list, dict, set, tuple
  • Complex objects: PartitionsDefinition, custom classes, etc.
  • Functions: Callable objects for use as UDFs in Jinja2 templates
Invalid Signatures:
# ❌ Multiple parameters not allowed
@dg.template_var
def invalid_multiple_params(context: ComponentLoadContext, other_param: str):
    pass

# ❌ Wrong context type
@dg.template_var
def invalid_context_type(context: ComponentDeclLoadContext):
    pass

# ❌ Static methods with parameters other than context
class MyComponent(dg.Component):
    @staticmethod
    @dg.template_var
    def invalid_static(param: str):  # Only 0 or 1 (context) params allowed
        pass

Examples:

Basic template variable (no context needed):

import dagster as dg
import os

@dg.template_var
def database_url() -> str:
    if os.getenv("ENVIRONMENT") == "prod":
        return "postgresql://prod-server:5432/db"
    else:
        return "postgresql://localhost:5432/dev_db"

Context-aware template variable:

@dg.template_var
def component_specific_table(context: dg.ComponentLoadContext) -> str:
    return f"table_{context.path.name}"

Template variable returning a function:

This is colloquially called a “udf” (user-defined function).

@dg.template_var
def table_name_generator() -> Callable[[str], str]:
    return lambda prefix: f"{prefix}_processed_data"

Using template variables in YAML:

# defs.yaml
type: my_project.components.DataProcessor
template_vars_module: .template_vars
attributes:
  database_url: "{{ database_url }}"
  table_name: "{{ component_specific_table }}"
  processed_table: "{{ table_name_generator('sales') }}"

Component class static methods:

class MyComponent(dg.Component):
    @staticmethod
    @dg.template_var
    def default_config() -> dict:
        return {"timeout": 30, "retries": 3}

    @staticmethod
    @dg.template_var
    def context_aware_value(context: dg.ComponentLoadContext) -> str:
        return f"value_for_{context.path.name}"

Using in YAML (component static methods):

type: my_project.components.MyComponent
attributes:
  config: "{{ default_config }}"
  name: "{{ context_aware_value }}"

See also: - dagster.ComponentLoadContext: Context object available to template variables

class dagster.Component [source]

Abstract base class for creating Dagster components.

Components are the primary building blocks for programmatically creating Dagster definitions. They enable building multiple interrelated definitions for specific use cases, provide schema-based configuration, and built-in scaffolding support to simplify component instantiation in projects. Components are automatically discovered by Dagster tooling and can be instantiated from YAML configuration files or Python code that conform to the declared schema.

Key Capabilities:

  • Definition Factory: Creates Dagster assets, jobs, schedules, and other definitions
  • Schema-Based Configuration: Optional parameterization via YAML or Python objects
  • Scaffolding Support: Custom project structure generation via dg scaffold commands
  • Tool Integration: Automatic discovery by Dagster CLI and UI tools
  • Testing Utilities: Built-in methods for testing component behavior

Implementing a component:

  • Every component must implement the build_defs() method, which serves as a factory for creating Dagster definitions.
  • Components can optionally inherit from Resolvable to add schema-based configuration capabilities, enabling parameterization through YAML files or structured Python objects.
  • Components can attach a custom scaffolder with the @scaffold_with decorator.

Examples:

Simple component with hardcoded definitions:

import dagster as dg

class SimpleDataComponent(dg.Component):
    """Component that creates a toy, hardcoded data processing asset."""

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        @dg.asset
        def raw_data():
            return [1, 2, 3, 4, 5]

        @dg.asset
        def processed_data(raw_data):
            return [x * 2 for x in raw_data]

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

Configurable component with schema:

import dagster as dg
from typing import List

class DatabaseTableComponent(dg.Component, dg.Resolvable, dg.Model):
    """Component for creating assets from database tables."""

    table_name: str
    columns: List[str]
    database_url: str = "postgresql://localhost/mydb"

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        @dg.asset(key=f"{self.table_name}_data")
        def table_asset():
            # Use self.table_name, self.columns, etc.
            return execute_query(f"SELECT {', '.join(self.columns)} FROM {self.table_name}")

        return dg.Definitions(assets=[table_asset])

Using the component in a YAML file (defs.yaml):

type: my_project.components.DatabaseTableComponent
attributes:
  table_name: "users"
  columns: ["id", "name", "email"]
  database_url: "postgresql://prod-db/analytics"

Component Discovery:

Components are automatically discovered by Dagster tooling when defined in modules specified in your project’s pyproject.toml registry configuration:

[tool.dagster]
module_name = "my_project"
registry_modules = ["my_project.components"]

This enables CLI commands like:

dg list components # List all available components in the Python environment
dg scaffold defs MyComponent path/to/component  # Generate component instance with scaffolding

Schema and Configuration:

To make a component configurable, inherit from both Component and Resolvable, along with a model base class. Pydantic models and dataclasses are supported largely so that pre-existing code can be used as schema without having to modify it. We recommend using dg.Model for new components, which wraps Pydantic with Dagster defaults for better developer experience.

  • dg.Model: Recommended for new components (wraps Pydantic with Dagster defaults)
  • pydantic.BaseModel: Direct Pydantic usage
  • @dataclass: Python dataclasses with validation

Custom Scaffolding:

Components can provide custom scaffolding behavior using the @scaffold_with decorator:

import textwrap

import dagster as dg
from dagster.components import Scaffolder, ScaffoldRequest


class DatabaseComponentScaffolder(Scaffolder):
    def scaffold(self, request: ScaffoldRequest) -> None:
        # Create component directory
        component_dir = request.target_path
        component_dir.mkdir(parents=True, exist_ok=True)

        # Generate defs.yaml with template
        defs_file = component_dir / "defs.yaml"
        defs_file.write_text(
            textwrap.dedent(
                f'''
            type: {request.type_name}
            attributes:
                table_name: "example_table"
                columns: ["id", "name"]
                database_url: "${{DATABASE_URL}}"
        '''.strip()
            )
        )

        # Generate SQL query template
        sql_file = component_dir / "query.sql"
        sql_file.write_text("SELECT * FROM example_table;")


@dg.scaffold_with(DatabaseComponentScaffolder)
class DatabaseTableComponent(dg.Component, dg.Resolvable, dg.Model):
    table_name: str
    columns: list[str]

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        # Component implementation
        pass

See also: - dagster.Definitions: The object returned by build_defs()

class dagster.StateBackedComponent [source]

Base class for components that depend on external state that needs to be fetched and cached.

State-backed components are designed for integrations where Dagster definitions depend on information from external systems (like APIs or compiled artifacts) rather than just code and configuration files. The component framework manages the lifecycle of fetching, storing, and loading this state.

Subclasses must implement:

  • write_state_to_path: Fetches state from external sources and writes it to a local path
  • build_defs_from_state: Builds Dagster definitions from the cached state
  • defs_state_config: Property that returns configuration for state management

Example:

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

import dagster as dg
from dagster.components import DefsStateConfig, DefsStateConfigArgs, ResolvedDefsStateConfig

@dataclass
class MyStateBackedComponent(dg.StateBackedComponent):
    base_url: str
    defs_state: ResolvedDefsStateConfig = DefsStateConfigArgs.local_filesystem()

    @property
    def defs_state_config(self) -> DefsStateConfig:
        return DefsStateConfig.from_args(
            self.defs_state, default_key=f"MyComponent[{self.base_url}]"
        )

    def write_state_to_path(self, state_path: Path) -> None:
        # Fetch table metadata from external API
        response = requests.get(f"{self.base_url}/api/tables")
        tables = response.json()
        # Write state to file as JSON
        state_path.write_text(json.dumps(tables))

    def build_defs_from_state(
        self, context: dg.ComponentLoadContext, state_path: Optional[Path]
    ) -> dg.Definitions:
        if state_path is None:
            return dg.Definitions()

        # Read cached state
        tables = json.loads(state_path.read_text())

        # Create one asset per table found in the state
        assets = []
        for table in tables:
            @dg.asset(key=dg.AssetKey(table["name"]))
            def table_asset():
                # Fetch and return the actual table data
                return fetch_table_data(table["name"])

            assets.append(table_asset)

        return dg.Definitions(assets=assets)

YAML configuration:

# defs.yaml
type: my_package.MyStateBackedComponent
attributes:
  base_url: "{{ env.MY_API_URL }}"
  defs_state:
    management_type: LOCAL_FILESYSTEM
class dagster.Resolvable [source]

Base class for making a class resolvable from yaml.

This framework is designed to allow complex nested objects to be resolved from yaml documents. This allows for a single class to be instantiated from either yaml or python without limiting the types of fields that can exist on the python class.

Key Features:

  • Automatic yaml schema derivation: A pydantic model is automatically generated from the class definition using its fields or init arguments and their annotations.
  • Jinja template resolution: Fields in the yaml document may be templated strings, which are rendered from the available scope and may be arbitrary python objects.
  • Customizable resolution behavior: Each field can customize how it is resolved from the yaml document using a :py:class:~dagster.Resolver.

Resolvable subclasses must be one of the following:

  • pydantic model
  • @dataclass
  • plain class with an annotated init
  • @record

Example:

import datetime
from typing import Annotated

import dagster as dg


def resolve_timestamp(
    context: dg.ResolutionContext,
    raw_timestamp: str,
) -> datetime.datetime:
    return datetime.datetime.fromisoformat(
        context.resolve_value(raw_timestamp, as_type=str),
    )


# the yaml field will be a string, which is then parsed into a datetime object
ResolvedTimestamp = Annotated[
    datetime.datetime,
    dg.Resolver(resolve_timestamp, model_field_type=str),
]


class MyClass(dg.Resolvable, dg.Model):
    event: str
    start_timestamp: ResolvedTimestamp
    end_timestamp: ResolvedTimestamp


# python instantiation
in_python = MyClass(
    event="test",
    start_timestamp=datetime.datetime(2021, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc),
    end_timestamp=datetime.datetime(2021, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc),
)

# yaml instantiation
in_yaml = MyClass.resolve_from_yaml(
    '''
event: test
start_timestamp: '{{ start_year }}-01-01T00:00:00Z'
end_timestamp: '{{ end_timestamp }}'
''',
    scope={
        # string templating
        "start_year": "2021",
        # object templating
        "end_timestamp": in_python.end_timestamp,
    },
)

assert in_python == in_yaml
class dagster.ResolutionContext [source]

The context available to Resolver functions when “resolving” from yaml in to a Resolvable object. This class should not be instantiated directly.

Provides a resolve_value method that can be used to resolve templated values in a nested object before being transformed into the final Resolvable object. This is typically invoked inside a Resolver’s resolve_fn to ensure that jinja-templated values are turned into their respective python types using the available template variables.

Example:

import datetime
import dagster as dg

def resolve_timestamp(
    context: dg.ResolutionContext,
    raw_timestamp: str,
) -> datetime.datetime:
    return datetime.datetime.fromisoformat(
        context.resolve_value(raw_timestamp, as_type=str),
    )
resolve_value [source]

Recursively resolves templated values in a nested object. This is typically invoked inside a Resolver’s resolve_fn to resolve all nested template values in the input object.

Parameters:

  • val (Any) – The value to resolve.
  • as_type (Optional[type]) – If provided, the type to cast the resolved value to. Used purely for type hinting and does not impact runtime behavior.

Returns: The input value after all nested template values have been resolved.

class dagster.Resolver [source]

Contains information on how to resolve a value from YAML into the corresponding Resolved class field.

You can attach a resolver to a field’s type annotation to control how the value is resolved.

Example:

import datetime
from typing import Annotated
import dagster as dg

def resolve_timestamp(
    context: dg.ResolutionContext,
    raw_timestamp: str,
) -> datetime.datetime:
    return datetime.datetime.fromisoformat(
        context.resolve_value(raw_timestamp, as_type=str),
    )

class MyClass(dg.Resolvable, dg.Model):
    event: str
    # the yaml field will be a string, which is then parsed into a datetime object
    timestamp: Annotated[
        datetime.datetime,
        dg.Resolver(resolve_timestamp, model_field_type=str),
    ]
class dagster.Model [source]

pydantic BaseModel configured with recommended default settings for use with the Resolved framework.

Extra fields are disallowed when instantiating this model to help catch errors earlier.

Example:

import dagster as dg

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

# raises exception
MyModel(name="John", age=30, other="field")

Core Models​

These Annotated TypeAliases can be used when defining custom Components for common Dagster types.

dagster.ResolvedAssetKey: Annotated[AssetKey, ...``] [source]

Allows resolving to an AssetKey via a YAML-friendly schema.

dagster.ResolvedAssetSpec: Annotated[AssetSpec, ...``] [source]

Allows resolving to an AssetSpec via a YAML-friendly schema.

dagster.AssetAttributesModel

A pydantic modeling of all the attributes of an AssetSpec that can be set before the definition is created.

dagster.ResolvedAssetCheckSpec: Annotated[AssetCheckSpec, ...``] [source]

Allows resolving to an AssetCheckSpec via a YAML-friendly schema.

Built-in Components​

class dagster.DefsFolderComponent [source]

A component that represents a directory containing multiple Dagster definition modules.

DefsFolderComponent serves as a container for organizing and managing multiple subcomponents within a folder structure. It automatically discovers and loads components from subdirectories and files, enabling hierarchical organization of Dagster definitions. This component also supports post-processing capabilities to modify metadata and properties of definitions created by its child components.

Key Features:

  • Post-Processing: Allows modification of child component definitions via configuration
  • Automatic Discovery: Recursively finds and loads components from subdirectories
  • Hierarchical Organization: Enables nested folder structures for complex projects

The component automatically scans its directory for:

  • YAML component definitions (defs.yaml files)
  • Python modules containing Dagster definitions
  • Nested subdirectories containing more components

Here is how a DefsFolderComponent is used in a project by the framework, along with other framework-defined classes.

my_project/
└── defs/
    ├── analytics/             # DefsFolderComponent
    │   ├── defs.yaml          # Post-processing configuration
    │   ├── user_metrics/      # User-defined component
    │   │   └── defs.yaml
    │   └── sales_reports/     # User-defined component
    │       └── defs.yaml
    └── data_ingestion/        # DefsFolderComponent
        ├── api_sources/       # DefsFolderComponent
        │   └── some_defs.py   # PythonFileComponent
        └── file_sources/      # DefsFolderComponent
            └── files.py       # PythonFileComponent

Parameters:

  • path – The filesystem path to the directory containing child components.
  • children – A mapping of child paths to their corresponding Component instances. This is typically populated automatically during component discovery.

DefsFolderComponent supports post-processing through its defs.yaml configuration, allowing you to modify definitions created by child components using target selectors

Examples:

Using post-processing in a folder’s defs.yaml:

# analytics/defs.yaml
type: dagster.DefsFolderComponent
post_processing:
  assets:
    - target: "*" # add a top level tag to all assets in the folder
      attributes:
        tags:
          top_level_tag: "true"
    - target: "tag:defs_tag=true" # add a tag to all assets in the folder with the tag "defs_tag"
      attributes:
        tags:
          new_tag: "true"

Please see documentation on post processing and the selection syntax for more examples.

Component Discovery:

The component automatically discovers children using these patterns:

  1. YAML Components: Subdirectories with defs.yaml files
  2. Python Modules: Any .py files containing Dagster definitions
  3. Nested Folders: Subdirectories that contain any of the above Files and directories matching these patterns are ignored:
  • __pycache__ directories
  • Hidden directories (starting with .)

Note: DefsFolderComponent instances are typically created automatically by Dagster’s component loading system. Manual instantiation is rarely needed unless building custom loading logic or testing scenarios.

When used with post-processing, the folder’s defs.yaml should only contain post-processing configuration, not component type definitions.

Loading Components​

dagster.load_from_defs_folder [source]

Constructs a Definitions object by automatically discovering and loading all Dagster definitions from a project’s defs folder structure.

This function serves as the primary entry point for loading definitions in dg-managed projects. It reads the project configuration (dg.toml or pyproject.toml), identifies the defs module, and recursively loads all components, assets, jobs, and other Dagster definitions from the project structure.

The function automatically handles:

  • Reading project configuration to determine the defs module location
  • Importing and traversing the defs module hierarchy
  • Loading component definitions and merging them into a unified Definitions object
  • Enriching definitions with plugin component metadata from entry points

Parameters: path_within_project (Path) – A path within the dg project directory. This directory or a parent of should contain the project’s configuration file (dg.toml or pyproject.toml with [tool.dg] section).Returns: A merged Definitions object containing all discovered definitions from the project’s defs folder, enriched with component metadata.

Return type: Definitions Example:

from pathlib import Path
import dagster as dg

@dg.definitions
def defs():
    project_path = Path("/path/to/my/dg/project")
    return dg.load_from_defs_folder(project_root=project_path)