Destinations¶

A destination persists an asset's data and reads it back for downstream assets.

destination ¶

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

Create a Destination subclass from a decorated class.

Relations can be declared via annotations:

@destination
class MyDest:
    connection: PostgresConnection

    def read(self, context): ...
    def write(self, context, data): ...

Or through the relations channel:

@destination(name="My destination", relations={"connection": PostgresConnection})
class MyDest:
    def read(self, context): ...
    def write(self, context, data): ...

An annotation naming a component class declares a relation rather than a Pydantic model field. relations maps a relation name to a Relation, a component class, or a list of component classes narrowing the relation the decorated class declares under that name; explicit declarations win over the class annotations. Every other keyword is one of the public ClassVars and field defaults the decorated class declares, or Destination itself for a plain class (key, name, icon, tags). An unknown name is a TypeError at decoration.

Destination ¶

Destination(**data: Any)

Bases: interloper.component.base.Component

A component that reads and writes asset data, one partition at a time.

A destination stores data per partition, None standing for the whole of an unpartitioned asset. The contract is three hooks: write_partition and read_partition for a single partition, and partition_row_counts for the coverage view. write and read own the rest, splitting a window write into one call per partition and joining a window read into one table, so a destination is partition-correct by construction. A backend that does not store per partition (a database that clears each partition and inserts a window in one batch) overrides write or read instead. Two more hooks, reference and create_view, serve views; a destination that holds no views leaves them refusing.

The hooks may be plain sync methods (the common case: most warehouse and file clients are sync) or async def for native async I/O. The engine is async-native: it awaits async implementations directly and offloads sync ones to a worker thread, so a destination never blocks the event loop either way. An annotation naming a component class declares a relation, which the destination resolves by name:

class JSONDestination(Destination):
    connection: BucketConnection

    def write_partition(self, context: IOContext, partition: Partition | None, data: Any) -> None:
        self.connection.put(self._path(context, partition), json.dumps(data, default=str))

    def read_partition(self, context: IOContext, partition: Partition | None) -> Any:
        return json.loads(self.connection.get(self._path(context, partition)))

key ¶

key = 'destination'

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

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'>)}

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 = {}

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.

create_view ¶

create_view(asset: Asset, definition: Any) -> None

Materialize a view's definition under the asset's table.

Parameters:

Name	Type	Description	Default
`asset`	`Asset`	The view being materialized; its `table` and `dataset` name the view.	required
`definition`	`Any`	Whatever the view's `data()` returned, in this destination's own terms.	required

Raises:

Type	Description
`ConfigError`	Always; this destination holds no views.

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(context: IOContext) -> dict[str, int]

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

The partition column is read from context.asset.partitioning.column. Each key in the returned dict is the string representation of a partition value; each value is the number of rows in that partition.

Parameters:

Name	Type	Description	Default
`context`	`IOContext`	Destination context (uses `asset.table`, `asset.dataset`, and `asset.partitioning`).	required

Returns:

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

read ¶

read(context: IOContext) -> Any

Read the data the context covers, a window of tables as one table.

Every partition in context.partitions is read through read_partition; several are joined through the data's representation, so a window comes back as one table in whatever type the destination holds natively. Data no representation recognises (objects a file destination pickled) cannot be joined, and comes back as one value per partition.

Parameters:

Name	Type	Description	Default
`context`	`IOContext`	IO context carrying the target asset, the window and the effective schema.	required

Returns:

Type	Description
`Any`	One table, whatever the window; a list of the partitions' values, in window order, for
`Any`	a window of non-tabular data.

read_partition ¶

read_partition(
    context: IOContext, partition: Partition | None
) -> Any

Load one partition.

Parameters:

Name	Type	Description	Default
`context`	`IOContext`	IO context carrying the target asset and the effective schema.	required
`partition`	`Partition \| None`	The partition to load, or `None` for the unpartitioned whole.	required

reference ¶

reference(asset: Asset) -> Any

This destination's handle on an asset's data, for a view to build on.

What a reference is depends on the destination: a SQL warehouse hands out a table reference, and a destination that holds no views refuses, as this default does.

Parameters:

Name	Type	Description	Default
`asset`	`Asset`	The asset a view wants to build on.	required

Raises:

Type	Description
`ConfigError`	Always; this destination holds no views.

write ¶

write(context: IOContext, data: Any) -> None

Write data, one partition at a time.

A window is split into one write_partition call per partition, each receiving its slice of the data; a single partition or the unpartitioned whole is one call receiving the data as is.

Parameters:

Name	Type	Description	Default
`context`	`IOContext`	IO context carrying the target asset, the window, and the effective schema.	required
`data`	`Any`	The data to write, in its native representation.	required

write_partition ¶

write_partition(
    context: IOContext,
    partition: Partition | None,
    data: Any,
) -> None

Store data for one partition.

Parameters:

Name	Type	Description	Default
`context`	`IOContext`	IO context carrying the target asset and the effective schema.	required
`partition`	`Partition \| None`	The partition being stored, or `None` for the unpartitioned whole.	required
`data`	`Any`	The partition's slice of the data to store.	required

IOContext ¶

IOContext(asset: Asset, window: PartitionWindow | None = None, metadata: dict[str, Any] = <factory>, schema: type[Schema] | None = None)

Immutable context passed to Destination.read and Destination.write.

Carries the target asset, the window the call covers and the run metadata, so a destination resolves the storage location, plus the effective schema of the data: the asset's declared schema when set, otherwise the one inferred during conform, None when none could be resolved. Destinations use it for DDL, typed loads and restoring types on read.

A destination stores data per partition, None standing for the unpartitioned whole; partitions and slices spell that out, so no destination branches on the window itself.

partitions ¶

partitions

The partitions this context covers, first to last.

Returns:

Type	Description
	`[None]` for the unpartitioned whole, else the window's partitions.

slices ¶

slices(data: Any) -> list[tuple[Partition | None, Any]]

Pair each partition with its slice of data.

A window of several partitions is split on the asset's partition column through the data's representation; a single partition, or the whole, receives the data as is.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The data being written, in its native representation.	required

Returns:

Type	Description
`list[tuple[Partition \| None, Any]]`	`(partition, slice)` pairs, one per partition, in `partitions` order.

Raises:

Type	Description
`ConfigError`	If a window of several partitions is written for an asset that declares no partitioning, since nothing says which column to slice on.

FileDestination ¶

Bases: interloper.destination.file.FileDestination

Destination that reads and writes pickle files on the local filesystem.

Data is stored under {base_path}/{dataset}/{table}/data.pkl (or {base_path}/{table}/data.pkl when no dataset is set). Partitioned assets add a {column}={key} subdirectory; the partition dispatch (including window splitting) is Destination's.

Unlike CSVDestination this stores whatever the asset returned, tabular or not — so a window write of a non-tabular object, which no representation can split, stores that whole object under each partition.

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)

name ¶

name = 'File'

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

relations ¶

relations = {}

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)

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

CSVDestination ¶

Bases: interloper.destination.csv.CSVDestination

Destination that reads and writes CSV files on the local filesystem.

Data is stored under {base_path}/{dataset}/{table}/data.csv (or {base_path}/{table}/data.csv when no dataset is set). Partitioned assets add a {column}={key} subdirectory; the partition dispatch (including window splitting) is Destination's.

Data is read as list[dict] rows through its registered representation on write (each dict is a row; the keys of the first dict determine the CSV headers).

CSV stores every value as a string. When the IO context carries a schema, reads reconcile the rows against it, restoring the declared types.

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)

name ¶

name = 'CSV'

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

relations ¶

relations = {}

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)

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

MemoryDestination ¶

Bases: interloper.destination.memory.MemoryDestination

Destination that stores data in a class-level dict keyed by {dataset}/{table}/{partition}.

All instances share a single _storage dict so data written by one asset is visible to others. The partition dispatch (including window splitting) is Destination's. Call clear between test runs.

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)

name ¶

name = 'Memory'

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

relations ¶

relations = {}

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)

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