Source code for dagster._core.definitions.asset_in

from typing import Any, Mapping, NamedTuple, Optional, Sequence, Type, Union

import dagster._check as check
from dagster._annotations import PublicAttr
from dagster._core.definitions.events import (
    AssetKey,
    CoercibleToAssetKey,
    CoercibleToAssetKeyPrefix,
)
from dagster._core.definitions.input import NoValueSentinel
from dagster._core.types.dagster_type import DagsterType, resolve_dagster_type
from dagster._utils.backcompat import canonicalize_backcompat_args

from .partition_mapping import PartitionMapping


[docs]class AssetIn( NamedTuple( "_AssetIn", [ ("key", PublicAttr[Optional[AssetKey]]), ("metadata", PublicAttr[Optional[Mapping[str, Any]]]), ("key_prefix", PublicAttr[Optional[Sequence[str]]]), ("input_manager_key", PublicAttr[Optional[str]]), ("partition_mapping", PublicAttr[Optional[PartitionMapping]]), ("dagster_type", PublicAttr[Union[DagsterType, Type[NoValueSentinel]]]), ], ) ): """ Defines an asset dependency. Attributes: key_prefix (Optional[Union[str, Sequence[str]]]): If provided, the asset's key is the concatenation of the key_prefix and the input name. Only one of the "key_prefix" and "key" arguments should be provided. key (Optional[Union[str, Sequence[str], AssetKey]]): The asset's key. Only one of the "key_prefix" and "key" arguments should be provided. metadata (Optional[Dict[str, Any]]): A dict of the metadata for the input. For example, if you only need a subset of columns from an upstream table, you could include that in metadata and the IO manager that loads the upstream table could use the metadata to determine which columns to load. partition_mapping (Optional[PartitionMapping]): Defines what partitions to depend on in the upstream asset. If not provided, defaults to the default partition mapping for the partitions definition, which is typically maps partition keys to the same partition keys in upstream assets. dagster_type (DagsterType): Allows specifying type validation functions that will be executed on the input of the decorated function before it runs. """ def __new__( cls, key: Optional[CoercibleToAssetKey] = None, metadata: Optional[Mapping[str, Any]] = None, key_prefix: Optional[CoercibleToAssetKeyPrefix] = None, asset_key: Optional[CoercibleToAssetKey] = None, input_manager_key: Optional[str] = None, partition_mapping: Optional[PartitionMapping] = None, dagster_type: Union[DagsterType, Type[NoValueSentinel]] = NoValueSentinel, ): key = canonicalize_backcompat_args(key, "key", asset_key, "asset_key", "1.0.0") if isinstance(key_prefix, str): key_prefix = [key_prefix] check.invariant( not (key and key_prefix), "key and key_prefix cannot both be set on AssetIn" ) return super(AssetIn, cls).__new__( cls, key=AssetKey.from_coerceable(key) if key is not None else None, metadata=check.opt_inst_param(metadata, "metadata", Mapping), key_prefix=check.opt_list_param(key_prefix, "key_prefix", of_type=str), input_manager_key=check.opt_str_param(input_manager_key, "input_manager_key"), partition_mapping=check.opt_inst_param( partition_mapping, "partition_mapping", PartitionMapping ), dagster_type=NoValueSentinel if dagster_type is NoValueSentinel else resolve_dagster_type(dagster_type), )