Assets¶

An asset is the unit of data in Interloper: something that produces data and can be materialized into destinations. This page covers how to define one, what it can receive, how to run it, and what happens between the function returning and the data landing.

Defining an asset¶

The usual form is a method on a source, decorated with @il.asset:

import interloper as il

@il.source
class Shop(il.Source):
    @il.asset
    def users(self) -> list[dict]:
        return fetch_users()

A standalone asset is a decorated function:

@il.asset
def users() -> list[dict]:
    return fetch_users()

And an asset can be a class, when it needs more than one method:

class Users(il.Asset):
    def data(self, **kwargs) -> list[dict]:
        return fetch_users()

All three produce an Asset definition (a class). Calling it creates an instance:

users_asset = users()
users_asset.key        # "users"

The key defaults to the snake_cased function or class name and can be overridden with key=. The docstring becomes the asset's description in the catalog.

Sync or async¶

The function may be sync or async def. Sync functions are offloaded to a worker thread; async functions are awaited on the event loop. Either way the engine never blocks:

@il.asset
async def events(self) -> list[dict]:
    return await self.connection.client.get("/events")

What an asset can return¶

Anything. Tabular data is what destinations and schemas are built for: a list[dict], or a pandas DataFrame when interloper-pandas is installed. As a convenience the asset unwraps a few Python shapes into rows before anything else sees them: a single dict becomes one row, pydantic models or a list of them are dumped, a generator of rows is consumed, and None is no rows. Non-tabular objects pass through untouched when the asset declares no schema, which is what a FileDestination storing pickles expects; with a schema they are an error.

Parameters¶

Every parameter of data() declares a relation, read off its annotation. self, context, source and **kwargs are reserved and declare nothing.

context: a parameter named context receives an ExecutionContext.

Resources: a parameter annotated with a component class (a config, a connection, or your own) receives whatever is bound to it, or the fallback that class can fill itself with. See Resources.

Upstreams: a parameter annotated il.Upstream receives an upstream asset and the data read from it, list[il.Upstream] a many-valued one. The parameter name is the bare asset key it expects, so inside a source it names a sibling; relations= declares anything else. See Dependencies.

@il.asset(relations={"raw": il.Relation("asset", "warehouse.raw_orders")})
def orders(
    self,
    context: il.ExecutionContext,      # the run's context
    connection: ShopConnection,        # a resource, by annotation
    users: il.Upstream,                # a sibling asset, by name
    raw: il.Upstream,                  # a cross-source asset, by relations=
) -> list[dict]:
    ...

A parameter annotated with anything else is a TypeError at class creation: nothing could ever fill it.

Decorator options¶

@il.asset(
    key="ads_stats",                                   # override the derived key
    name="Ads statistics",                             # display name
    icon="carbon:chart-line",                          # icon identifier for UIs
    tags=["Report"],                                   # catalog tags
    schema=AdsStats,                                   # output schema
    partitioning=il.TimePartitionConfig(column="date"),
    relations={                                        # the relation channel, keyed by parameter
        "destinations": [il.CSVDestination],           # allowed destination classes
        "connection": il.Relation(AdsConnection),
        "campaigns": il.Relation("asset", "ads.campaigns"),
        "budget": il.Relation("asset", "finance.budget", optional=True),
    },
    materialization_strategy=il.MaterializationStrategy.RECONCILE,
    normalizer=il.Normalizer(flatten_max_level=1),
)
def ads_stats(
    self,
    context: il.ExecutionContext,
    connection: AdsConnection,
    campaigns: il.Upstream,
    budget: il.Upstream | None = None,
) -> list[dict]:
    ...

The three channels are described in Decorator options. A list of classes under relations["destinations"] restricts the destination classes an asset accepts; instances are supplied at construction.

Instance configuration¶

An instance carries the runtime state a definition does not know about:

Its relations carry the rest: connection, config, destinations, every upstream. They are constructor keywords named after the relation, and bind, unbind and assignment change them afterwards. See Dependencies.

Set them at construction, or derive a reconfigured copy by calling an existing instance:

asset = users(destinations=il.CSVDestination(base_path="./data"), dataset="shop")

read_only = asset(enabled=False)
strict = asset(materialization_strategy=il.MaterializationStrategy.STRICT)
bare = asset(normalizer=None)           # unset: the source's normalizer, or none

Naming a field or a relation changes it, leaving it out keeps it; None clears a relation and makes a setting the asset inherits from its source inherit again. An asset owned by a source reads its dataset, default_destination_key, normalizer, materialization_strategy, retry and unbound relations through the source on every read (see Sources), and runs only while the source is enabled too.

asset.default_destination is the destination downstream assets read it from: the one whose key is default_destination_key, else the first.

rewired = asset(connection=OtherConnection(...), campaigns=[fb.campaigns, tt.campaigns])

Unknown keyword arguments raise TypeError rather than being silently dropped.

Running and materializing¶

window is required for partitioned assets and ignored for unpartitioned ones; a bare partition is accepted as the window of one. dag is required when the asset has a bound upstream, because upstream data is read through the DAG's own node. metadata is a free-form dict (run id, backfill id) carried onto every event the run emits.

An asset that produces no data skips its destination writes and emits a warning log event rather than writing an empty table.

What happens on materialize¶

1. The run's window is checked by the asset's partitioning.
2. Every relation is resolved: a resource to what is bound or its fallback, an upstream to an il.Upstream read from its destination.
3. data() runs, wrapped in asset_data_* events and a tracing span.
4. The normalizer reshapes the result, when one is configured.
5. The result is conformed to the schema according to the materialization strategy. Without a schema, one is inferred so destinations know the column types.
6. Each destination's write() is called with an IOContext, wrapped in dest_write_* events.

Steps 4 and 5 run off the event loop. Every step emits events and spans.

Execution context¶

An asset that declares a context parameter receives an ExecutionContext:

Each partition accessor raises an AttributeError explaining why it is unavailable: the asset is not partitioned, no window was given, the window covers several partitions, or the granularity is not daily. See Partitioning.

Identity¶

Row counts per partition¶

asset.partition_row_counts() asks the first destination for row counts grouped by the partition column. It raises when the asset is not partitioned or has no destination.