Implementation:Dagster io Dagster Asset Decorator

Knowledge Sources	Dagster Dagster Docs
Domains	Data_Engineering, Orchestration
Last Updated	2026-02-10 00:00 GMT

Overview

Concrete decorator for defining software-defined assets provided by the Dagster core library.

Description

The @asset decorator is the primary mechanism for defining software-defined assets in Dagster. It transforms a Python function into an AssetsDefinition object that the Dagster framework can schedule, execute, and track. The decorator accepts a wide range of parameters controlling asset identity, partitioning, grouping, automation, and metadata.

When applied to a function, the decorator inspects the function signature to determine upstream dependencies (via parameter names matching other asset keys or explicit deps/ins arguments). The function body becomes the computation that materializes the asset.

Usage

Import the decorator from the dagster package and apply it to any Python function that produces a data artifact. Use when defining individual assets in a Dagster pipeline. For multi-asset outputs, use @multi_asset instead.

Code Reference

Source Location

Repository: dagster
File: python_modules/dagster/dagster/_core/definitions/decorators/asset_decorator.py:L154

Signature

def asset(
    compute_fn: Optional[Callable[..., Any]] = None,
    *,
    name: Optional[str] = None,
    key_prefix: Optional[CoercibleToAssetKeyPrefix] = None,
    ins: Optional[Mapping[str, AssetIn]] = None,
    deps: Optional[Iterable[CoercibleToAssetDep]] = None,
    metadata: Optional[ArbitraryMetadataMapping] = None,
    tags: Optional[Mapping[str, str]] = None,
    description: Optional[str] = None,
    required_resource_keys: Optional[AbstractSet[str]] = None,
    io_manager_key: Optional[str] = None,
    partitions_def: Optional[PartitionsDefinition[str]] = None,
    op_tags: Optional[Mapping[str, Any]] = None,
    group_name: Optional[str] = None,
    output_required: bool = True,
    automation_condition: Optional[AutomationCondition[AssetKey]] = None,
    retry_policy: Optional[RetryPolicy] = None,
    code_version: Optional[str] = None,
    key: Optional[CoercibleToAssetKey] = None,
    check_specs: Optional[Sequence[AssetCheckSpec]] = None,
    owners: Optional[Sequence[str]] = None,
    kinds: Optional[AbstractSet[str]] = None,
    pool: Optional[str] = None,
) -> Union[AssetsDefinition, Callable[[Callable[..., Any]], AssetsDefinition]]

Import

from dagster import asset

# or

import dagster as dg

@dg.asset
def my_asset(): ...

I/O Contract

Inputs

Name	Type	Required	Description
compute_fn	Callable[..., Any]	No	The decorated function that computes the asset. Provided implicitly when using decorator syntax.
name	str	No	The name of the asset. Defaults to the function name.
key_prefix	CoercibleToAssetKeyPrefix	No	A prefix for the asset key, useful for namespacing.
ins	Mapping[str, AssetIn]	No	Explicit mapping of input parameter names to upstream asset specifications.
deps	Iterable[CoercibleToAssetDep]	No	Assets that must be materialized before this asset, without passing data.
metadata	ArbitraryMetadataMapping	No	Static metadata to attach to the asset definition.
tags	Mapping[str, str]	No	Tags for categorizing and filtering assets.
description	str	No	Human-readable description displayed in the Dagster UI.
partitions_def	PartitionsDefinition[str]	No	Partitioning scheme for the asset (e.g., daily, monthly, static).
group_name	str	No	Logical group name for organizing assets in the UI.
automation_condition	AutomationCondition[AssetKey]	No	Condition under which the asset should be automatically materialized.
kinds	AbstractSet[str]	No	Set of kind tags (e.g., "duckdb", "python") displayed as badges in the UI.
key	CoercibleToAssetKey	No	Full asset key, alternative to name + key_prefix.
pool	str	No	Concurrency pool for limiting parallel materializations.

Outputs

Name	Type	Description
return	AssetsDefinition	A fully configured asset definition object that can be included in a Dagster Definitions object.

Usage Examples

DuckDB Data Extraction

import dagster as dg

@dg.asset(kinds={"duckdb"}, group_name="ingestion", description="Load raw customer data")
def raw_customers(duckdb: DuckDBResource) -> None:
    with duckdb.get_connection() as conn:
        conn.execute(
            "CREATE TABLE IF NOT EXISTS raw_customers AS "
            "SELECT * FROM read_csv('https://example.com/customers.csv')"
        )

Asset with Dependencies

import dagster as dg

@dg.asset(deps=["raw_customers"], kinds={"duckdb", "python"}, group_name="transform")
def clean_customers(duckdb: DuckDBResource) -> dg.MaterializeResult:
    with duckdb.get_connection() as conn:
        result = conn.execute(
            "CREATE TABLE clean_customers AS "
            "SELECT * FROM raw_customers WHERE customer_id IS NOT NULL"
        )
        count = conn.execute("SELECT COUNT(*) FROM clean_customers").fetchone()[0]
    return dg.MaterializeResult(metadata={"row_count": count})

Related Pages

Implements Principle

Principle:Dagster_io_Dagster_Software_Defined_Assets

Requires Environment

Uses Heuristic

Heuristic:Dagster_io_Dagster_Mandatory_Ruff_Formatting

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment