Dependencies¶

An asset can consume the output of other assets. Inside a DAG, each upstream is materialized first, read back from its destination, and handed to the downstream asset as an il.Upstream: the upstream asset plus the data read for the run's scope.

An upstream is one relation among the others a component declares (a connection, a config, a destination). One primitive, il.Relation, declares them all; only the ones whose kind is asset become edges in the graph. See Resources for the rest.

Inside a source¶

Annotate a parameter il.Upstream and name it after a sibling asset. The source binds it when it builds its assets:

import interloper as il

@il.source
class Shop(il.Source):
    @il.asset(partitioning=il.TimePartitionConfig(column="date"))
    def orders(self, context: il.ExecutionContext) -> list[dict]:
        return [{"date": context.partition_date, "id": 1, "total": 9.99}]

    @il.asset(partitioning=il.TimePartitionConfig(column="date"))
    def order_stats(self, context: il.ExecutionContext, orders: il.Upstream) -> list[dict]:
        rows = orders.data or []
        return [{"date": context.partition_date, "count": len(rows)}]

The parameter name is the relation name and the bare asset key it expects, so orders: il.Upstream declares il.Relation("asset", "orders"), a sibling of the same source instance. A None default makes the relation optional, meaning its wiring may be absent, and declares on_delete="detach" with it: a parameter that tolerates a missing leg has no claim on the upstream, so deleting that upstream is allowed and simply drops the wiring, where a required upstream refuses the deletion while it is bound.

    @il.asset(partitioning=il.TimePartitionConfig(column="date"))
    def report(
        self,
        context: il.ExecutionContext,
        orders: il.Upstream,
        refunds: il.Upstream | None = None,
    ) -> list[dict]:
        ...

self, context, source and **kwargs are reserved. Any other parameter that is neither an il.Upstream nor a component class is a TypeError at class creation: nothing could ever fill it.

Declaring relations explicitly¶

When the parameter name is not the upstream's key, or the upstream lives in another source, write the relation on the decorator. relations= is keyed by parameter name and wins over the annotation:

@il.source
class Finance(il.Source):
    currency: str = il.InputField(default="EUR")

    @il.asset(
        partitioning=il.TimePartitionConfig(column="date"),
        relations={
            "orders": il.Relation("asset", "shop.orders"),
            "fx": il.Relation("asset", "rates.daily_fx", optional=True),
        },
    )
    def revenue(
        self,
        context: il.ExecutionContext,
        orders: il.Upstream,
        fx: il.Upstream | None,
    ) -> list[dict]:
        ...

A class-based asset writes the same relation as a typed class attribute:

class Revenue(il.Asset):
    orders: il.Asset = il.Relation("asset", "shop.orders")
    partitioning = il.TimePartitionConfig(column="date")

    def data(self, context: il.ExecutionContext, orders: il.Upstream) -> list[dict]:
        ...

The annotation is for the type checker; the relation itself says what it accepts. Keys come in four forms:

Many upstreams¶

many=True binds every matching asset at once, and data() receives a list[il.Upstream]:

@il.source(tags=["Analytics"])
class CampaignMatcher(il.Source):
    @il.asset(
        partitioning=il.TimePartitionConfig(column="date"),
        tags=["Entity"],
        relations={"campaigns": il.Relation("asset", "*.dim_campaigns", many=True)},
    )
    def campaign_matches(
        self,
        context: il.ExecutionContext,
        campaigns: list[il.Upstream],
    ) -> list[dict]:
        return [
            {"date": context.partition_date, **row}
            for leg in campaigns
            if leg.data is not None
            for row in leg.data
        ]

The DAG binds every dim_campaigns asset it holds, so the set of legs is decided by wiring, not by the class. list[il.Upstream] on its own infers the same relation on the parameter's bare key, which is a many-valued sibling; relations= is what reaches outside the source.

Fan-in across sources¶

The relation above is not hypothetical: interloper_assets.CampaignMatcher declares exactly this wildcard (same campaign_matches decorator shown above, plus schema=schemas.CampaignMatches), matching campaigns across every advertising platform into one lookup table. It reads each platform's star schema rather than its connector: not every connector collects a campaigns entity, so a star schema's dim_campaigns is built from whatever carries the campaigns (a snapshot, a campaign report, the ad facts), and the matcher always finds them under one key. Each row names the campaign's platform and account, the normalised canonical_name it is matched on and the match_id shared by every campaign matched to it. A key_pattern on the source selects the part of a name that identifies the campaign under a naming convention, and a similarity_threshold below 1.0 also merges names that merely resemble each other.

A star schema fans in one level below, with a qualified key: FacebookAdsStarSchema declares facebook_ads.ads_stats with many=True on its fact_ad_performance, one leg per Facebook Ads account. The star schemas are placeholders for now: they declare their wiring but materialize no rows.

Building a DAG over the star schemas and the matcher wires every dim_campaigns asset held into the campaigns relation, one leg per platform:

fb = FacebookAdsStarSchema(...)   # declares a dim_campaigns asset
tt = TiktokAdsStarSchema(...)     # declares a dim_campaigns asset
matcher = CampaignMatcher(destinations=[dest])

dag = il.DAG(fb, tt, matcher)   # binds matcher.campaign_matches.campaigns to [fb.dim_campaigns, tt.dim_campaigns]

A matcher built against upstreams already materialized elsewhere reads them read-only instead of running them, the same explicit-bind pattern as any other relation:

matcher.campaign_matches.bind("campaigns", fb.dim_campaigns)
dag = il.DAG(matcher)   # fb.dim_campaigns joins the DAG as enabled=False

examples/campaign_matcher.yaml is the same wiring written as a manifest: two star schemas each override their dim_campaigns asset to enabled: false, and campaign_matches names both by {ref: ...} since an asset always travels under its own source. See Specs for the manifest format.

How wiring works¶

Wiring is always an instance bound to a relation, never an id. It happens in three places:

• At construction, for siblings: Source.sibling_bindings() says which of an asset's relations resolve inside the source, and the source binds them to the instances it just built, right after building them.
• In the DAG, for everything a bare key cannot reach: a qualified or wildcard key is matched against the DAG's other assets. A many relation binds every candidate, a single-valued one the only candidate there is. Several candidates for a single-valued relation raise DAGError and ask to be bound by hand.
• By hand, at any point:

dest = il.MemoryDestination()
shop = Shop(destinations=[dest])
fin = Finance(destinations=[dest])

dag = il.DAG(shop, fin)          # binds fin.revenue.orders to shop.orders
fin.revenue.bound("orders")      # the shop.orders instance

fin.revenue.bind("orders", shop.orders)     # explicit; accumulates for a many relation
fin.revenue.orders = shop.orders            # assignment replaces atomically
fin.revenue.unbind("orders", shop.orders)   # refused if it would empty a non-optional relation

Binding goes through bind() everywhere, so an explicit binding is never overwritten by the DAG, and a hydrated one is never overwritten by construction.

In a spec, a bound upstream is a {ref: id} under the relation's name, since an asset always travels inside its own source's document. See Specs.

Reading upstream data¶

Each leg is read from the upstream's default_destination_key destination, or its first, scoped to the window the upstream consumes, joined into one table when it covers several partitions. il.Upstream carries the asset and its data, plus one accessor:

A view's upstreams are not read at all: data holds the destination's reference to the upstream instead. See Views & references.

data is None, with a LOG warning naming the leg, when the upstream has nothing materialized where the destination looks: no table or object for that scope at all (MemoryDestination and the file destinations key storage per partition, so a missing partition is exactly this; BigQueryDestination reaches it only when the table itself does not exist). An existing but empty scope is not this case, and comes back as whatever the destination returns for an empty read, an empty list or frame. optional governs wiring only, never data: the asset decides what a missing leg means. Any other read failure fails the asset.

A bound upstream the run does not hold at all is skipped with its own warning, so its leg simply does not exist.

Reads emit dest_read_* events and an interloper.destination.read span.

Rules the DAG enforces¶

validate_relations(nodes) is the check behind the three ConfigError rows. Constructing a component binds what it is given and checks each binding, but not that every required relation is bound: components are wired piecewise (a source's assets read its bindings through it, a manifest binds its references once every component exists), so the check runs where the graph is whole. The DAG runs it with its own nodes; loading a manifest runs it without nodes on each root, which is what lets a relation only the graph can fill stay unbound until then.

A non-partitioned asset never depends on a partitioned one because a partitioned upstream is read one partition at a time, which an unpartitioned downstream cannot express. A run has one partition scope, so both ends of an edge must agree on granularity, read-only upstreams included.

Running one asset with its parents¶

A bound upstream the run does not materialize joins the DAG anyway, as a disabled copy under the same id: it is a live instance with its own destinations, so it can be read without being run. That is the whole mechanism behind running one asset with its parents:

mini = dag.mini_dag(fin.revenue.id)
[(op.qualified_key, op.enabled) for op in mini.operations]
# [("finance.revenue", True), ("shop.orders", False)]

Only the target executes; the parents are read, not rewritten. Partition checks still apply along those edges. A source offers the same idea over its own assets through select.