Skip to content

Sources

The Source step, and the shorthands that build one. A source's methodology is a location, which it reads rows through and content-addresses them. It is where data enters a plan.

matchlab.sources.sources

Source — the leaf of a plan.

A source reads rows from a Location and content-addresses them. It takes no inputs, so it is where raw data, and therefore non-determinism, enters a plan. Its spec key includes a hash of the data it read, which is what makes a freshly constructed Source pick up changes at the origin while an existing object memoises its read.

Classes:

  • Source –

    A location's rows, read and content-addressed.

Functions:

Source 

Source(location_class: type[Location] | str, name: str, location_settings: dict[str, Any] | None = None, location_resources: dict[str, Any] | None = None, key_field: str = 'id')

Bases: RecordStep


              flowchart TD
              matchlab.sources.sources.Source[Source]
              matchlab.recordstep.RecordStep[RecordStep]
              matchlab.steps.Step[Step]

                              matchlab.recordstep.RecordStep --> matchlab.sources.sources.Source
                                matchlab.steps.Step --> matchlab.recordstep.RecordStep
                



              click matchlab.sources.sources.Source href "" "matchlab.sources.sources.Source"
              click matchlab.recordstep.RecordStep href "" "matchlab.recordstep.RecordStep"
              click matchlab.steps.Step href "" "matchlab.steps.Step"

A location's rows, read and content-addressed.

A source is a RecordStep. Read directly, its records are its extract, with id set to each row's content-addressed leaf, so a model can match over it with no intervening step.

Define a source.

Parameters:

  • location_class

    (type[Location] | str) –

    A Location subclass, or its registered name.

  • name

    (str) –

    The source's name within the plan. Prefixes every column this source contributes, so it must be usable in a SQL identifier.

  • location_settings

    (dict[str, Any] | None, default: None ) –

    That location's configuration — the query, for a RelationalDB. Serialisable, carried in a document, and hashed into this source's fingerprint.

  • location_resources

    (dict[str, Any] | None, default: None ) –

    Resources the location needs that cannot be serialised, keyed by field name, such as {"client": Resource("warehouse", engine)}. See matchlab.resources.

  • key_field

    (str, default: 'id' ) –

    The name of the unique identifier field. Read as a string whatever the location returns it as.

Raises:

  • ValueError –

    If the name could not prefix a SQL identifier, or if key_field is not a column name.

  • ResourceError –

    If a field was passed in the wrong one of location_settings and location_resources, or if one resource name covers two different objects.

Methods:

  • f –

    Prefix one or more field names with this source's name.

  • fetch –

    Read from the location and yield the resulting rows in batches.

  • sample –

    Peek at the first n rows without collecting.

  • leaves –

    Return (key, leaf), each source key mapped to its leaf cluster.

  • collect –

    Materialise this step and everything it depends on.

  • lineage –

    Return this step and all its inputs, upstream-first.

  • draw –

    Render this step's sub-plan as a tree.

  • fingerprints –

    Address every artifact this plan is made of, its own and its inputs'.

  • identifiers –

    Return (id, source, key, leaf) for every record this record step reads.

  • data –

    Return this record step's records, collecting the plan first if needed.

  • transform –

    Reshape this record step with a transformer.

  • select –

    Keep only the named columns, plus id.

  • clean –

    Derive columns with DuckDB SQL, keeping the rest.

  • group –

    Collapse each id to one row using aggregate SQL.

  • dedupe –

    Deduplicate this record step.

  • link –

    Link this record step to another. A Source is one, needing no wrapping.

Attributes:

kind class-attribute 

kind: StepKind = SOURCE

location instance-attribute 

location: Location

location_class instance-attribute 

location_class: type[Location]

location_settings instance-attribute 

location_settings: dict[str, Any]

location_resources instance-attribute 

location_resources: dict[str, Resource]

name instance-attribute 

name: str

key_field instance-attribute 

key_field: str

parents property 

parents: tuple[Step, ...]

A source is a leaf in a plan.

spec property 

spec: SourceSpec

The serialisable spec for this source.

qualified_key property 

qualified_key: str

This source's key field, prefixed with the source name.

index_fields property 

index_fields: list[str]

Every column the extract returns except the key, in sorted order.

Read from the location rather than declared, so it cannot drift from what the location actually returns.

is_collected property 

is_collected: bool

Whether this step has been materialised.

f 

f(fields: str | Iterable[str]) -> str | list[str]

Prefix one or more field names with this source's name.

fetch 

fetch(qualify_names: bool = False, batch_size: int | None = None, return_type: DataFrameType = POLARS) -> Generator[DataFrameClass, None, None]

Read from the location and yield the resulting rows in batches.

sample 

sample(n: int = 100, return_type: DataFrameType = POLARS) -> DataFrameClass

Peek at the first n rows without collecting.

leaves 

leaves() -> DataFrame

Return (key, leaf), each source key mapped to its leaf cluster.

collect 

collect(store: Store | None = None, interactive: bool | None = None) -> Self

Materialise this step and everything it depends on.

Steps whose artifact is already stored are skipped without being run, so re-collecting after adding a downstream step only does the new work.

Reports as it goes: the plan, a record per step, and a closing summary of what ran, what was cached, how long it took and what the store now holds. No logging setup is needed for any of that — a collection lends the matchlab logger a console handler where the application hasn't configured one, and leaves an application that has entirely alone. See matchlab.core.logging.audible.

Parameters:

  • store
    (Store | None, default: None ) –

    Where to read and write artifacts. Defaults to the module-level store (a DuckDB store in the user cache directory).

  • interactive
    (bool | None, default: None ) –

    Whether someone is watching. None, the default, takes a terminal or a notebook as a yes. When they are, the plan is drawn as a live tree redrawn in place, and not logged. The tree on screen is the key those [step N] records need, and it stays there. When they are not, the plan is logged instead. See matchlab.progress.

Returns:

  • Self –

    This step, now collected.

lineage 

lineage() -> list[Step]

Return this step and all its inputs, upstream-first.

draw 

draw() -> str

Render this step's sub-plan as a tree.

fingerprints 

fingerprints() -> set[Fingerprint]

Address every artifact this plan is made of, its own and its inputs'.

Which artifacts belong to a plan is the plan's own business, so this is where a store gets told: store.prune(keep=plan.fingerprints()) hands storage a set of addresses it already understands, rather than a graph it would have to learn to walk.

Returns:

  • set[Fingerprint] –

    One fingerprint per step in lineage(). A set, because two steps in one

  • set[Fingerprint] –

    plan can address the same artifact. Identical specs over identical

  • set[Fingerprint] –

    inputs is the same bytes, and it is stored once.

Raises:

  • RuntimeError –

    If any step has not been collected. An uncollected plan names no artifacts at all, so answering with a smaller set would quietly tell a caller that less is worth keeping than they think.

identifiers 

identifiers(store: Store) -> DataFrame

Return (id, source, key, leaf) for every record this record step reads.

id is the resolver's entity root when reading through one, otherwise the source leaf. This is the upstream resolver output a downstream resolver needs to carry every reachable leaf forward, including records no model matched.

data 

data(return_type: DataFrameType = POLARS) -> DataFrameClass

Return this record step's records, collecting the plan first if needed.

transform 

transform(transformer: Transformer | type[Transformer] | str, transformer_settings: dict | None = None, transformer_resources: dict | None = None) -> Transform

Reshape this record step with a transformer.

select 

select(*columns: str) -> Transform

Keep only the named columns, plus id.

clean 

clean(cleaning: dict[str, SQLExpression]) -> Transform

Derive columns with DuckDB SQL, keeping the rest.

group 

group(aggregates: dict[str, SQLExpression]) -> Transform

Collapse each id to one row using aggregate SQL.

dedupe 

dedupe(model_class: type[Deduper] | str, model_settings: dict | None = None, model_resources: dict | None = None) -> Model

Deduplicate this record step.

link(other: RecordStep, model_class: type[Linker] | str, model_settings: dict | None = None, model_resources: dict | None = None) -> Model

Link this record step to another. A Source is one, needing no wrapping.

add_location_class 

add_location_class(location_class: type[Location]) -> None

Register a custom location so it can be named in a plan document.

resolve_location_class 

resolve_location_class(location_class: str | type[Location]) -> type[Location]

Return a location class, looking a name up in the registry.

How Source accepts either the class or its registered name, and how matchlab.document rebuilds the locations a plan reads.

Raises:

  • ValueError –

    If no location class of that name is registered here.

read_database 

read_database(name: str, *, sql: SQLQuery, client: DBClient | Resource[DBClient], key_field: str = 'id') -> Source

Read a source from a relational database.

Parameters:

  • name

    (str) –

    The source's name within the plan.

  • sql

    (SQLQuery) –

    The query producing the rows. Every column other than the key contributes to record identity. Run as written, so trust it: see RelationalDB.

  • client

    (DBClient | Resource[DBClient]) –

    A SQLAlchemy engine or ADBC connection, or a Resource naming one. Name it to dump the plan; share the same Resource between sources reading one warehouse.

  • key_field

    (str, default: 'id' ) –

    The unique identifier column.

Returns:

  • Source –

    The source, uncollected.

read_dataframe 

read_dataframe(name: str, *, df: DataFrameClass | Resource[DataFrameClass], key_field: str = 'id') -> Source

Read a source from a dataframe already in memory.

Parameters:

  • name

    (str) –

    The source's name within the plan.

  • df

    (DataFrameClass | Resource[DataFrameClass]) –

    A polars, pandas or arrow frame, or a Resource naming one. Every column other than the key contributes to record identity, so shape the frame before handing it over.

  • key_field

    (str, default: 'id' ) –

    The unique identifier column.

Returns:

  • Source –

    The source, uncollected.

matchlab.sources.base

The contract every location implements.

A Location is a methodology, exactly as a Deduper or a Transformer is. A Source names the class and hands it settings and resources separately, so the query travels in a document and the client never does. See matchlab.resources.

Classes:

  • Location –

    A place data is read from, holding everything needed to read it.

Location

Bases: BaseModel, ABC


              flowchart TD
              matchlab.sources.base.Location[Location]

              

              click matchlab.sources.base.Location href "" "matchlab.sources.base.Location"

A place data is read from, holding everything needed to read it.

Every field is a setting unless marked matchlab.resources.FromResources.

A location declares no version, unlike the methodologies the other kinds of step run. A source hashes the rows it read into its own spec key, so a change in what a location returns already moves the fingerprint.

Methods:

  • read –

    Get this location's rows, in batches.

Attributes:

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', frozen=True, arbitrary_types_allowed=True)

read abstractmethod 

read(batch_size: int | None = None, rename: dict[str, str] | Callable | None = None, return_type: DataFrameType = POLARS, schema_overrides: dict[str, DataType] | None = None) -> Iterator[DataFrameClass]

Get this location's rows, in batches.

Parameters:

  • batch_size
    (int | None, default: None ) –

    The size used for internal batching.

  • rename
    (dict[str, str] | Callable | None, default: None ) –

    Renaming to apply to the rows that come back.

    • If a dictionary is provided, it will be used to rename the columns.
    • If a callable is provided, it will take the old name as input and return the new name.
  • return_type
    (DataFrameType, default: POLARS ) –

    The type of data to return. Defaults to "polars".

  • schema_overrides
    (dict[str, DataType] | None, default: None ) –

    Types to force on the columns that come back, rather than letting the location infer them.