Documentation · Tutorial · Changelog
Assets are functions that produce data. Sources group them, destinations store them, a DAG runs them. The same code runs in a notebook, a container, or the scheduled platform.
uv add interloper-coreA connection holds credentials and a client. It is a pydantic-settings model: values come from
the constructor, .env, or the environment (SHOP_API_KEY here).
from functools import cached_property
import interloper as il
from pydantic_settings import SettingsConfigDict
@il.connection(name="Shop API")
class ShopConnection(il.Connection):
model_config = SettingsConfigDict(env_prefix="shop_")
api_key: str = il.SecretField()
@cached_property
def client(self) -> il.RESTClient:
return il.RESTClient("https://api.shop.example", auth=il.HTTPBearerAuth(self.api_key))A source groups assets. Configuration fields and the connection live on the class; assets are
methods and read both through self. A parameter annotated il.Upstream and named after a
sibling asset is a dependency, and a schema types the data on write and on read-back.
import datetime as dt
class Order(il.Schema):
id: int
total: float
class OrderStats(il.Schema):
date: dt.date
orders: int
revenue: float | None
@il.source(tags=["Commerce"])
class Shop(il.Source):
connection: ShopConnection
account: str = il.InputField(description="Shop account id", discriminator=True)
@il.asset(schema=Order)
def orders(self) -> list[dict]:
rows: list[dict] = []
paginator = il.PageNumberPaginator(total_path="meta.pages")
for page in self.connection.client.paginate("/orders", paginator, data_selector="data"):
rows.extend(page)
return rows
@il.asset(schema=OrderStats, partitioning=il.TimePartitionConfig(column="date"), tags=["Report"])
def order_stats(self, context: il.ExecutionContext, orders: il.Upstream) -> list[dict]:
day = context.partition_date # also: context.partition, .window, .logger, .metadata
rows = orders.data or [] # what `orders` wrote, read back from its destination
context.logger.info(f"{len(rows)} orders for {self.account}")
return [{"date": day, "orders": len(rows), "revenue": sum(o["total"] for o in rows)}]Instances carry the runtime configuration: resources, destinations, dataset. Destinations
cascade from the source to its assets. discriminator=True on account makes the tables
orders__acme and order_stats__acme, so several accounts share one dataset.
shop = Shop(
account="acme",
connection=ShopConnection(api_key="..."), # omit to load from the environment
destinations=[il.CSVDestination(base_path="./data")], # built in: CSV, pickle, memory; BigQuery and GCS via interloper-google-cloud
)Your own destination is two methods, sync or async, each handling one partition, or the
unpartitioned whole. Windows are split and gathered for you; DatabaseDestination adds the
delete-then-insert dance for tables.
import json
from pathlib import Path
@il.destination
class JSONLDestination(il.Destination):
base_path: str = ""
def _path(self, context: il.IOContext, partition: il.Partition | None) -> Path:
base = Path(self.base_path) / context.asset.dataset / context.asset.table
return base / ("data.jsonl" if partition is None else f"{partition.id}.jsonl")
def write_partition(self, context: il.IOContext, partition: il.Partition | None, data) -> None:
path = self._path(context, partition)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(json.dumps(row, default=str) for row in data))
def read_partition(self, context: il.IOContext, partition: il.Partition | None):
return [json.loads(line) for line in self._path(context, partition).read_text().splitlines()]Run a single asset, or build a DAG. The DAG validates the wiring (missing dependencies, cycles, a non-partitioned asset downstream of a partitioned one) and runs assets in dependency order. Partitioned assets always run for a partition; a window is a loop.
shop.orders.run() # execute and return the data, write nothing
shop.orders.materialize() # execute and write to every destination
dag = il.DAG(shop)
dag.materialize(il.TimePartition(dt.date(2026, 1, 15))) # default AsyncRunner
# RunResult(status=completed, partition=2026-01-15, completed=2, failed=0, canceled=0, time=0.01s)
for partition in il.TimePartitionWindow(dt.date(2026, 1, 1), dt.date(2026, 1, 7)): # newest first
dag.materialize(partition)
runner = il.AsyncRunner(max_workers=8, fail_fast=False, on_event=print) # or SerialRunner, MultiProcessRunner
result = il.run(runner.run(dag, il.TimePartition(dt.date(2026, 1, 15)))) # il.run: sync bridge, notebook-safe
result.status, result.failed_ids, result.executionsEvery component serializes to a spec and back, so a run can be described in YAML. ${VAR} is
read from the environment. The runner comes from interloper.yaml or INTERLOPER_RUNNER_*.
# shop.yaml
path: shop.Shop
init:
account: acme
resources:
connection:
path: shop.ShopConnection
init:
api_key: ${SHOP_API_KEY}
destinations:
- path: interloper.destination.csv.CSVDestination
init: { base_path: ./data }interloper run -f shop.yaml --date 2026-01-15 --dry-run
interloper run -f shop.yaml --date 2026-01-15
interloper run -f shop.yaml --date 2026-01 # monthly key for monthly assets; also 2026, 2026-01-15T13Components describe themselves. A package registers its components with one entry point and
they appear in the catalog the API, the UI and spec key references read.
Shop.definition().config_schema # JSON Schema of the configuration fields
shop.to_spec() # the YAML above, as data
il.Catalog.discover() # every component installed packages declare[project.entry-points."interloper.components"]
shop = "shop"The documentation has a page per concept, an extension guide (component model, representations, runners, operations) and a reference section.
| Package | Provides |
|---|---|
interloper-core |
The framework |
interloper-pandas |
pandas DataFrame representation and normalizer |
interloper-google-cloud |
Google Cloud connection, BigQuery and GCS destinations |
interloper-slack |
Slack connection and notification hook |
interloper-assets |
Ready-made sources for advertising, analytics and commerce platforms |
interloper-docker |
Docker runner and launcher |
interloper-k8s |
Kubernetes runner and launcher |
interloper-db |
Persistence: components, relations, runs, events, migrations |
interloper-scheduler |
Cron, hooks, credential renewal, queue worker, reaper |
interloper-api |
FastAPI backend |
interloper-app |
Web UI (Nuxt SPA) |
interloper-mcp |
MCP server over the catalog, lineage and run history |
interloper-agent |
AI agent (Google ADK) |
interloper-toolkit |
Read-only tool functions shared by the agent and MCP server |
One version for all packages, released together to PyPI.
interloper app runs the API, cron controller, queue worker and reaper against Postgres,
configured by interloper.yaml and INTERLOPER_* variables. Images are on
GHCR as
interloper-<role>:<version> for api, scheduler, core, mcp and frontend, each
(bar the frontend) in two variants: the bare tag is loaded, carrying the role's packages
plus every component class a catalog can name, and -slim is pure, core plus the role's
own packages, meant to be extended. The
Helm chart sits alongside them as an OCI artifact. See RELEASING.md and
Running the app.
helm install interloper oci://ghcr.io/digitl-cloud/charts/interloper --version <version>make setup # pre-commit hooks + uv sync --all-packages --all-extras
make check # ruff, ty, pytest, frontend lint and typecheck
make dev # seeded local instance with the web UI on :3000
uv run zensical serve # documentation siteLayout, conventions and the local dev instance are in AGENTS.md.
Apache 2.0. See LICENSE.