Assets¶

An asset is a unit of data: a function or class that produces it, the context it runs in, and the upstream handles it reads other assets through.

asset ¶

asset(target: Any = None, /, **settings: Any) -> Any

Create an Asset subclass from a decorated function.

The decorated function may be sync or async. Sync functions are offloaded to a worker thread at materialization time; async functions are awaited natively on the event loop:

@asset
def users(**kwargs):
    return fetch_users()

@asset
async def events(**kwargs):
    return await fetch_events()

@asset(tags=["Report"], schema=UsersStats, relations={"destinations": [BigQueryDestination]})
def users_stats(config: MyConfig, connection: MyConn) -> Any:
    return fetch_stats()

Every parameter of the decorated function declares a relation, inferred from its annotation; relations is for the declarations an annotation cannot express, such as a cross-source or many-valued upstream. It maps a data() parameter name to a Relation, a component class (the shorthand for a relation on it), or a list of component classes narrowing the relation Asset declares under that name; explicit declarations win over the inferred ones.

Every other keyword is definition metadata or behaviour: the public ClassVars and field defaults Asset declares (key, name, icon, tags, schema, partitioning, dataset, normalizer, materialization_strategy, ...). An unknown name is a TypeError at decoration. The result is an Asset subclass with the function as its data().

Asset ¶

Asset(**data: Any)

Bases: interloper.operation.base.Operation

A data-producing component.

Subclass and implement data() to define an asset. Every parameter of that signature is filled at run time: context and source are injected, and every other parameter declares a relation read off its annotation. A component class is filled with whatever is bound to it, and il.Upstream (or list[il.Upstream]) with the data read from an upstream asset of the parameter's name:

class Revenue(Asset):
    def data(self, connection: MyConnection, orders: Upstream) -> Any:
        return connection.price(orders.data)

Or use the @asset decorator for a functional style:

@asset
def revenue(connection: MyConnection, orders: Upstream) -> Any:
    return connection.price(orders.data)

dataset ¶

dataset = <interloper.component.inherited.InheritedSetting object at 0x7febf2404350>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

default_destination ¶

default_destination

The destination downstream readers load this asset from.

The destination whose key is default_destination_key when one is set and bound, else the first bound destination.

Raises:

Type	Description
`AssetError`	If the asset has no destination at all.

default_destination_key ¶

default_destination_key = <interloper.component.inherited.InheritedSetting object at 0x7febf24c5160>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

destinations ¶

destinations = Relation(
    kind="destination",
    key="",
    many=True,
    optional=True,
    default=None,
    on_delete="block",
    name="destinations",
)

One declared link from an owner component to the components that may fill it.

kind and key name the acceptable targets: kind names one or several component kinds, key (when non-empty) narrows to specific keys within those kinds, matched through ComponentIdentity.satisfies. many marks a relation that binds several components at once; optional marks one that may stay unbound or be left empty, and says nothing else: on_delete alone decides what deleting a bound target does, block refusing the deletion while this relation holds the target and detach letting it through and dropping the binding. default and has_fallback together describe a relation that can be resolved without an explicit binding.

A relation is also its own descriptor: Component._collect installs the stamped copy under the relation's name, so the class attribute reads as the declaration and the instance attribute as what is bound to it, and assignment rebinds it.

The declaration form the framework writes pairs the relation with the annotation of what fills it (destinations: list[Destination] = Relation("destination", many=True)). The TYPE_CHECKING __new__ is what makes that form check: the type checker reads a Pydantic field of the annotated type, so the constructor kwarg and the instance attribute are both typed, while the runtime sees a Relation (Pydantic ignores it, and Component._collect drops the annotation before fields are collected).

Two attributes are not declared but derived. name is stamped from the attribute the relation is declared under, and target is the class the Relation(cls) shorthand was written with, the class a fallback is built from; neither is a constructor argument. default (intended type Callable[[], Component] | None) and target (intended type type[Component] | None) are typed Any for the type checker, since Component is only imported under TYPE_CHECKING; both stay out of dumps.

enabled ¶

enabled = <interloper.component.inherited.InheritedSetting object at 0x7febf28ca0f0>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

internal_fields ¶

internal_fields = frozenset({'normalizer'})

frozenset() -> empty frozenset object frozenset(iterable) -> frozenset object

Build an immutable unordered collection of unique elements.

key ¶

key = 'asset'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

kind ¶

kind = 'asset'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

materialization_strategy ¶

materialization_strategy = <interloper.component.inherited.InheritedSetting object at 0x7febf25beb70>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

model_config ¶

model_config = {'arbitrary_types_allowed': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>, <class 'interloper.component.relation.Relation'>)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

normalizer ¶

normalizer = <interloper.component.inherited.InheritedSetting object at 0x7febf436deb0>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

relations ¶

relations = {
    "destinations": Relation(
        kind="destination",
        key="",
        many=True,
        optional=True,
        default=None,
        on_delete="block",
        name="destinations",
    )
}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

retry ¶

retry = <interloper.component.inherited.InheritedSetting object at 0x7febf24c5550>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

source ¶

source

The source this asset belongs to, if any.

An asset's owning source is its parent (see parent); only a source ever parents an asset, so the cast is safe.

table ¶

table

The physical table (or leaf) name this asset materializes to, or refers to.

Named by the owning source (see Source.table); a standalone asset uses its class key.

tags ¶

tags = []

Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

body_from_function ¶

body_from_function(fn: FunctionType) -> ClassBody

Read a decorated function into an asset body: the function becomes data().

A function whose first parameter is self is a method asset: it receives the asset's source as self. The generated data() wraps the function in the functools.wraps sense, so relation inference reads the original's annotations through __wrapped__ and the stamped __signature__ is what the parameters read as; an async function gives an async data(), which the run awaits on its loop.

Parameters:

Name	Type	Description	Default
`fn`	`FunctionType`	The decorated function.	required

Returns:

Type	Description
`ClassBody`	The body carrying the generated `data()`.

classpath ¶

classpath() -> str

Fully qualified import path for this asset class.

Source-owned assets return the composite form "module:SourceName.AssetName", where the colon explicitly marks the module / attribute boundary. Resolution walks the attribute chain at class level: an asset class read on its source class is the class itself (see AssetMeta), with no instantiation required.

Standalone assets return the regular dotted module path.

Returns:

Type	Description
`str`	Import path string.

data ¶

data(**kwargs: Any) -> Any

Return this asset's data.

Subclasses must override this method.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	The resolved call arguments: the execution context, the owning source and whatever fills the declared relations, keyed by the parameter names of the overriding signature.	required

Raises:

Type	Description
`NotImplementedError`	If the subclass does not implement `data()`.

definition ¶

definition() -> AssetDefinition

Produce a structured definition of this asset class.

The path comes from classpath, so a source-owned asset gets the composite "module.Source:asset_kind" form.

Returns:

Type	Description
`AssetDefinition`	An AssetDefinition with the output schema and partitioning inlined.

execute ¶

execute(context: OperationContext) -> OperationResult

Materialize this asset: the asset's operation.

The runner-facing adapter over materialize_async; the manual entry points (run, materialize) stay the authoring API.

Parameters:

Name	Type	Description	Default
`context`	`OperationContext`	The facts this execution is scoped to.	required

Returns:

Type	Description
`OperationResult`	An effectless result: a materialization's effects are its
`OperationResult`	destination writes and the events it emits.

materialize ¶

materialize(
    window: Partition | PartitionWindow | None = None,
    dag: DAG | None = None,
    metadata: dict[str, Any] | None = None,
) -> Any

Execute the asset and write the result to all configured destinations.

Sync entrypoint for scripts, REPLs, and notebooks; drives materialize_async to completion on the bridge loop (see interloper.run):

asset.materialize()

Async code awaits materialize_async instead.

Parameters:

Name	Type	Description	Default
`window`	`Partition \| PartitionWindow \| None`	The window the run covers; a partition is accepted as the window of one.	`None`
`dag`	`DAG \| None`	DAG for dependency resolution (required if asset has upstreams).	`None`
`metadata`	`dict[str, Any] \| None`	Arbitrary metadata dict (e.g. run_id, backfill_id).	`None`

Returns:

Type	Description
`Any`	The execution result, or `None` if the asset is disabled.

materialize_async ¶

materialize_async(
    window: Partition | PartitionWindow | None = None,
    dag: DAG | None = None,
    metadata: dict[str, Any] | None = None,
) -> Any

Execute the asset and write the result to all configured destinations.

Parameters:

Name	Type	Description	Default
`window`	`Partition \| PartitionWindow \| None`	The window the run covers; a partition is accepted as the window of one.	`None`
`dag`	`DAG \| None`	DAG for dependency resolution (required if asset has upstreams).	`None`
`metadata`	`dict[str, Any] \| None`	Arbitrary metadata dict (e.g. run_id, backfill_id).	`None`

Returns:

Type	Description
`Any`	The execution result, or `None` if the asset is disabled.

model_post_init ¶

model_post_init(context: Any) -> None

Default id to a generated UUID if not provided.

Parameters:

Name	Type	Description	Default
`context`	`Any`	The pydantic validation context, unused here.	required

partition_row_counts ¶

partition_row_counts() -> dict[str, int]

Return row counts grouped by this asset's partition column.

Delegates to Destination.partition_row_counts using the configured default destination.

Returns:

Type	Description
`dict[str, int]`	Mapping from partition value (as string) to row count.

Raises:

Type	Description
`PartitionError`	If this asset is not partitioned.

run ¶

run(
    window: Partition | PartitionWindow | None = None,
    dag: DAG | None = None,
    metadata: dict[str, Any] | None = None,
) -> Any

Execute the asset and return the result without writing to destination.

Sync entrypoint for scripts, REPLs, and notebooks; drives run_async to completion on the bridge loop (see interloper.run):

data = asset.run()

Async code awaits run_async instead.

Parameters:

Name	Type	Description	Default
`window`	`Partition \| PartitionWindow \| None`	The window the run covers; a partition is accepted as the window of one.	`None`
`dag`	`DAG \| None`	DAG for dependency resolution (required if asset has upstreams).	`None`
`metadata`	`dict[str, Any] \| None`	Arbitrary metadata dict (e.g. run_id, backfill_id).	`None`

Returns:

Type	Description
`Any`	The raw execution result.

run_async ¶

run_async(
    window: Partition | PartitionWindow | None = None,
    dag: DAG | None = None,
    metadata: dict[str, Any] | None = None,
) -> Any

Execute the asset and return the result without writing to destination.

Resolves the context and the inputs, runs data() and shapes its result (see _produce); a sync data() is offloaded to a thread, an async one awaited.

Parameters:

Name	Type	Description	Default
`window`	`Partition \| PartitionWindow \| None`	The window the run covers; a partition is accepted as the window of one.	`None`
`dag`	`DAG \| None`	DAG for dependency resolution (required if asset has upstreams).	`None`
`metadata`	`dict[str, Any] \| None`	Arbitrary metadata dict (e.g. run_id, backfill_id).	`None`

Returns:

Type	Description
`Any`	The raw execution result.

AssetDefinition ¶

Bases: interloper.component.base.ComponentDefinition

Definition of an asset: what may fill its relations, its schema, its partitioning.

Cross-entity references travel as keys: relations names the kinds and keys that may fill each declared link. Same-entity data is inlined: asset_schema is the asset's own output schema and partitioning its own partition config.

Asset keys come in three forms:

Bare key, "campaigns": scoped to the parent source. Used for intra-source relations.
Qualified key, "facebook_ads.campaigns": globally unique. Used for cross-source relations.
Wildcard key, "*.campaigns": that asset key from any source. Used by many-valued relations to fan in across providers.

The qualified_key property returns the globally unique form.

model_config ¶

model_config = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

qualified_key ¶

qualified_key

Globally unique asset key: source_key.asset_key.

Falls back to the bare key if no source key is set (e.g. standalone assets not owned by a source).

ExecutionContext ¶

ExecutionContext(
    asset: Asset,
    window: PartitionWindow | None = None,
    metadata: dict[str, Any] | None = None,
)

What an asset's data() knows about the run it executes in.

Built by the asset for each run from itself, the run's window and the run metadata, and handed to data() as its context parameter. Each partition accessor raises an AttributeError naming why it is unavailable: the asset is not partitioned, no window was given, the window holds several partitions, or the granularity is not daily.

Bind the context to one run of one asset.

Parameters:

Name	Type	Description	Default
`asset`	`Asset`	The asset executing.	required
`window`	`PartitionWindow \| None`	The window the run covers, `None` for an unpartitioned asset.	`None`
`metadata`	`dict[str, Any] \| None`	Run metadata (run id, backfill id), carried onto every event the asset emits; `None` is an empty dict.	`None`

asset ¶

asset

The asset executing: its key, table, dataset, source and partitioning.

logger ¶

logger

Logger whose messages become LOG events attributed to the asset.

metadata ¶

metadata

The run metadata (run id, backfill id, anything the caller passed).

partition ¶

partition

The partition this run covers: the window's only one.

The partition answers everything about its own scope: id, bounds and granularity for a time partition, value for a list one.

Raises:

Type	Description
`AttributeError`	If the asset is not partitioned, no window was given, or the window covers several partitions.

partition_date ¶

partition_date

The partition's day, for daily time partitioning only.

The typed reading of context.partition.value: the one accessor that hands an asset a real date without narrowing a type first, and it asserts the daily granularity rather than assuming it.

Raises:

Type	Description
`AttributeError`	If the asset is not time-partitioned at the daily granularity, no window was given, or the window covers several partitions.

window ¶

window

The window this run covers, of one partition unless the asset allows windows.

Upstream ¶

Upstream(asset: Asset, data: Any)

One upstream asset as handed to data(), with what its destination holds for the run.

Attributes:

Name	Type	Description
`asset`		The upstream asset the data was read from; its `source`, `identity` and `id` tell the upstreams apart.
`data`		The data read from the upstream's destination for the run's partition, in whatever representation that destination holds natively (rows from a file destination, a DataFrame from BigQuery). `None` when the upstream has nothing materialized where the destination looks (no table or object for that partition at all); a `LOG` warning names the upstream. An existing but empty partition is not this case: it arrives as whatever the destination returns for an empty read (an empty list or frame), not `None`. For a view, the destination's reference to the upstream (a `TableReference` on a SQL warehouse) rather than data: a view reads nothing.

rows ¶

rows

The data as rows, whatever representation the destination handed back.

The one line a consumer needs to be indifferent to where its upstream is stored: a DataFrame and a list of rows both come out as list[dict], and an upstream with nothing materialized comes out empty.

Returns:

Type	Description
	One mapping per row; `[]` when `data` is `None`.

read ¶

read(
    upstream: Asset,
    reader: Asset,
    window: PartitionWindow | None,
    metadata: dict[str, Any],
    *,
    parameter: str,
) -> Upstream

Read an upstream's data from its default destination for the window it consumes.

Nothing materialized where the destination looks is not a failure: the upstream is handed over with data=None and a warning event, and the reader decides what that means. Any other failure fails the reader.

Parameters:

Name	Type	Description	Default
`upstream`	`Asset`	The asset whose data is read.	required
`reader`	`Asset`	The asset reading it, whose identity the events carry.	required
`window`	`PartitionWindow \| None`	The run's window, narrowed here to what upstream consumes.	required
`metadata`	`dict[str, Any]`	Run-level metadata carried onto the emitted events.	required
`parameter`	`str`	The reader's `data()` parameter the upstream is bound to, which the missing-data warning names.	required

Returns:

Type	Description
`Upstream`	The upstream with its data.

Raises:

Type	Description
`AssetError`	If the read fails for any reason other than missing data.

reference ¶

reference(
    upstream: Asset, destination: Destination
) -> Upstream

Refer to an upstream's table instead of reading it, for a kind that reads no data.

Parameters:

Name	Type	Description	Default
`upstream`	`Asset`	The asset referred to.	required
`destination`	`Destination`	The destination whose handle on the upstream is wanted.	required

Returns:

Type	Description
`Upstream`	The upstream with the destination's reference as its data.