Sources¶

A source owns a set of assets and the connection, config and destinations they share.

source ¶

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

Create a Source subclass from a decorated class or function.

A decorated class declares its configuration fields, its assets (methods carrying @asset) and its helpers; a decorated function returns the asset classes and turns its own annotated parameters into configuration fields:

@source(tags=["Advertising"], dataset="raw_shop", relations={"connection": ShopConnection})
class Shop(Source): ...

relations maps a relation name to a Relation, a component class (the shorthand for a relation on it), or a list of component classes narrowing the relation Source declares under that name; explicit declarations win over the ones read from the class annotations.

Every other keyword is definition metadata or behaviour: the public ClassVars and field defaults Source declares (key, name, icon, tags, dataset, default_destination_key, normalizer, materialization_strategy, ...). An unknown name is a TypeError at decoration. The result is a Source subclass with its assets discovered.

Source ¶

Source(**data: Any)

Bases: interloper.component.base.Component, interloper.operation.base.Workload

A grouping component that holds assets with shared relations and destinations.

Define a source by subclassing: an annotation naming a component class declares a relation the source's assets inherit, and asset_types (or asset classes written in the body) names what it materializes:

class MySource(Source):
    connection: MyConnection
    asset_types = [Users, Orders]

An asset is reached by its key: on the class it is the asset class, on an instance the live asset (see AssetMeta):

MySource.users      # the asset class
MySource().users    # this source's Asset with key "users"

Assets written in the class body are the typed form; one given only in asset_types is reached the same way at runtime.

asset_types ¶

asset_types = []

Built-in mutable sequence.

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

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.

internal_fields ¶

internal_fields = frozenset(
    {"assets", "normalizer", "enabled"}
)

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

Build an immutable unordered collection of unique elements.

key ¶

key = 'source'

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 = 'source'

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'.

model_config ¶

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

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)

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)

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 a source body: what it returns are the assets.

The function is called once, for the asset classes it returns (one, a list, or nothing for a source without assets); its annotated parameters become configuration fields, their defaults the field defaults.

Parameters:

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

Returns:

Type	Description
`ClassBody`	The body carrying `asset_types` and the fields.

Raises:

Type	Description
`TypeError`	If the function returned anything but nothing, an asset class or a list of them, naming what it returned.

definition ¶

definition() -> SourceDefinition

Produce a structured definition of this source including its assets.

An asset shows the less mature of its own maturity and the source's: an alpha source makes every one of its assets alpha. Both are read from the validated definitions, never from the declared class values.

Returns:

Type	Description
`SourceDefinition`	A SourceDefinition with metadata and nested asset definitions.

format ¶

format(template: str) -> str

Fill a location template with this source's fields.

An asset's dataset and a reference's table may name a field of the source in braces ("analytics_{property_id}"), which is how a declaration written once in the class body varies per instance.

Parameters:

Name	Type	Description	Default
`template`	`str`	A table or dataset name, possibly with `{field}` placeholders.	required

Returns:

Type	Description
`str`	The name with every placeholder filled.

Raises:

Type	Description
`ConfigError`	If the template names something that is not a field of this source.

inherited ¶

inherited(child: Component, name: str, own: Any) -> Any

What one of this source's assets reads for an inherited setting.

The dataset is part of an asset's location, which the source names (as table names the rest): the asset's own dataset, else this source's, else this source's key, with any {field} filled from this source's fields by format. Every other setting but enabled is handed down as Component.inherited does.

An asset runs only while it and this source are both enabled, which is how source(enabled=False) puts a read-only source in a run.

Parameters:

Name	Type	Description	Default
`child`	`Component`	The asset reading the setting.	required
`name`	`str`	The setting's field name.	required
`own`	`Any`	The asset's stored value, `None` while unset.	required

Returns:

Type	Description
`Any`	The effective value of the setting for child.

model_post_init ¶

model_post_init(context: Any) -> None

Build the source's assets, own them, check where they land, and wire them to each other.

The assets are incomplete when built here: this source's own relation kwargs, which they read through it, are bound only after model_post_init returns. That is why nothing is validated at construction; validate_relations cascades into each asset when the DAG or reconstruction asks, with everything this source can fill already bound.

Parameters:

Name	Type	Description	Default
`context`	`Any`	Pydantic's post-init context, forwarded untouched to `super().model_post_init`.	required

Raises:

Type	Description
`ConfigError`	If an asset already belongs to another source.

operations ¶

operations() -> list[Operation]

The source's own assets: the operations a run on it executes.

Returns:

Type	Description
`list[Operation]`	The assets this source holds.

sibling_bindings ¶

sibling_bindings() -> dict[str, dict[str, str]]

Which of this source's assets each asset relation resolves to, by key.

Only intra-source wiring is decided here: a declared key that resolves to another source, or to a wildcard, needs the whole DAG to be resolved and is left to it. Public for the relation store, which persists the same wiring; connector code never needs it.

Returns:

Type	Description
`dict[str, dict[str, str]]`	Asset key to a map of relation name to the sibling asset key that
`dict[str, dict[str, str]]`	fills it; assets with no sibling wiring are absent.

table ¶

table(asset: Asset) -> str

The physical table name of one of this source's assets.

A reference's table is its table_name as written, with this source's fields filled in by format. Any other asset's is the asset key suffixed with the instance's discriminator (the config field marked discriminator=True) and coerced to a valid identifier, so instances of a multi-account source materialize side by side in one dataset instead of overwriting each other's data. Without a discriminator the asset key is used as-is.

Override for another composition; keep the {asset.key}__{suffix} shape so tables stay wildcard-queryable per asset, leave references to super(), and return a valid identifier: the result is validated when the source resolves its assets, not coerced.

Parameters:

Name	Type	Description	Default
`asset`	`Asset`	The asset to name a table for.	required

Returns:

Type	Description
`str`	The physical table name for the asset.

validate_relations ¶

validate_relations(
    nodes: Mapping[str, Component] | None = None,
) -> None

Check this source's own relations, then cascade into every materializing asset.

An asset this source does not materialize is only ever read, so what fills its own relations is nothing any run needs: it is left unchecked, the same position _check_relations takes on a node it only reads. That is what keeps a source reconstructible from a partial set of assets, where a read-only asset names a sibling the document does not carry.

Parameters:

Name	Type	Description	Default
`nodes`	`Mapping[str, Component] \| None`	Every node materializing in the same run, keyed by id, forwarded to each asset's own check; see `validate_relations`. `None` skips the DAG-membership check.	`None`

SourceDefinition ¶

Bases: interloper.component.base.ComponentDefinition

Definition of a source including its nested asset definitions.

Cross-entity references use keys: relations names the kinds and keys that may fill each declared link. Same-entity data is inlined: assets are owned by this source, so their definitions are nested.

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)

partitionings ¶

partitionings() -> list[dict[str, Any]]

The partitioning of each partitioned asset the source declares.

Returns:

Type	Description
`list[dict[str, Any]]`	One partitioning per partitioned asset, in declaration order.