Skip to content

DAGs

matchbox.client.dags

Objects to define a DAG which indexes, deduplicates and links data.

Classes:

  • DAGNodeExecutionStatus –

    Enumeration of node execution statuses.

  • DAG –

    Self-sufficient pipeline of indexing, deduping and linking steps.

Attributes:

DAGExecutionStatus module-attribute 

DAGExecutionStatus: TypeAlias = dict[str, DAGNodeExecutionStatus]

CACHE_DIR module-attribute 

CACHE_DIR = user_cache_path('matchbox')

DAGNodeExecutionStatus

Bases: StrEnum


              flowchart TD
              matchbox.client.dags.DAGNodeExecutionStatus[DAGNodeExecutionStatus]

              

              click matchbox.client.dags.DAGNodeExecutionStatus href "" "matchbox.client.dags.DAGNodeExecutionStatus"

Enumeration of node execution statuses.

Attributes:

SKIPPED class-attribute instance-attribute 

SKIPPED = 'skipped'

DONE class-attribute instance-attribute 

DONE = 'done'

DOING class-attribute instance-attribute 

DOING = 'doing'

DAG 

DAG(name: CollectionName, admin_group: GroupName = PUBLIC)

Self-sufficient pipeline of indexing, deduping and linking steps.

Parameters:

  • name

    (CollectionName) –

    The name of the DAG, and therefore the collection it will connect to

  • admin_group

    (GroupName, default: PUBLIC ) –

    The name of the group that will be given admin permission over the DAG. Defaults to public, where anyone can modify, delete or run it

Methods:

  • list_all –

    List available DAG names on the server.

  • source –

    Create Source and add it to the DAG.

  • model –

    Create Model and add it to the DAG.

  • resolver –

    Create a resolver and add it to the DAG.

  • add_step –

    Add a step to the DAG.

  • get_source –

    Get a source by name from the DAG.

  • get_model –

    Get a model by name from the DAG.

  • get_resolver –

    Get a resolver by name from the DAG.

  • query –

    Create Query object.

  • draw –

    Create a string representation of the DAG.

  • new_run –

    Start a new run.

  • set_client –

    Assign a client to all sources at once.

  • load_default –

    Attach to default run in this collection, loading all DAG nodes.

  • load_pending –

    Attach to the pending run in this collection, loading all DAG nodes.

  • run_and_sync –

    Run entire DAG and send results to server.

  • set_default –

    Set the current run as the default for the collection.

  • lookup_key –

    Matches IDs against the selected backend.

  • get_matches –

    Return ResolverMatches, optionally filtering.

Attributes:

name instance-attribute 

name: CollectionName = CollectionName(name)

admin_group instance-attribute 

admin_group: GroupName = GroupName(admin_group)

nodes instance-attribute 

nodes: dict[StepName, Source | Model | Resolver] = {}

graph instance-attribute 

graph: dict[StepName, list[StepName]] = {}

run property writable 

run: RunID

Return run ID if available, else error.

sequence property 

sequence: list[StepName]

Return nodes in topological execution order.

Returns:

  • list[StepName] –

    List of node names in the order they would be executed by run_and_sync.

  • list[StepName] –

    Use as start/finish values to control partial execution.

final_steps property 

final_steps: list[Source | Model | Resolver]

Returns all apex nodes in the DAG.

Returns:

default_resolver property 

default_resolver: Resolver

Return the default resolver for this DAG.

list_all classmethod 

list_all() -> list[CollectionName]

List available DAG names on the server.

source 

source(*args: Any, **kwargs: Any) -> Source

Create Source and add it to the DAG.

model 

model(*args: Any, **kwargs: Any) -> Model

Create Model and add it to the DAG.

resolver 

resolver(*args: Any, **kwargs: Any) -> Resolver

Create a resolver and add it to the DAG.

add_step 

add_step(name: StepName, step: Step) -> None

Add a step to the DAG.

get_source 

get_source(name: StepName) -> Source

Get a source by name from the DAG.

Parameters:

  • name
    (StepName) –

    The name of the source to retrieve.

Returns:

  • Source –

    The Source object.

Raises:

  • ValueError –

    If the name doesn’t exist in the DAG or isn’t a Source.

get_model 

get_model(name: StepName) -> Model

Get a model by name from the DAG.

Parameters:

  • name
    (StepName) –

    The name of the model to retrieve.

Returns:

  • Model –

    The Model object.

Raises:

  • ValueError –

    If the name doesn’t exist in the DAG or isn’t a Model.

get_resolver 

get_resolver(name: StepName) -> Resolver

Get a resolver by name from the DAG.

query 

query(*args: Any, **kwargs: Any) -> Query

Create Query object.

draw 

draw(status: DAGExecutionStatus | None = None, mode: Literal['tree', 'list'] = 'tree') -> str

Create a string representation of the DAG.

In tree mode, nodes are shown in a dependency tree.

In list mode, nodes are shown in execution order as a numbered list.

If status is provided, it will show the status of each node. The status indicators are:

  • ✅ Done
  • 🔄 Working
  • ⏸️ Awaiting
  • ⏭️ Skipped

Node type indicators are:

  • 💎 Resolver
  • ⚙️ Model
  • 📄 Source

Parameters:

  • status
    (DAGExecutionStatus | None, default: None ) –

    Object describing the status of each node.

  • mode
    (Literal['tree', 'list'], default: 'tree' ) –

    “tree” renders the DAG as a tree structure (default). “list” renders nodes in flat execution order.

Returns:

  • str –

    String representation of the DAG with status indicators.

new_run 

new_run() -> Self

Start a new run.

set_client 

set_client(client: Any) -> Self

Assign a client to all sources at once.

load_default 

load_default() -> Self

Attach to default run in this collection, loading all DAG nodes.

load_pending 

load_pending() -> Self

Attach to the pending run in this collection, loading all DAG nodes.

Pending is defined as the last non-default run.

run_and_sync 

run_and_sync(start: str | None = None, finish: str | None = None, low_memory: bool = False, batch_size: int | None = None, profile: bool = False) -> None

Run entire DAG and send results to server.

Parameters:

  • start
    (str | None, default: None ) –

    Name of first node to run

  • finish
    (str | None, default: None ) –

    Name of last node to run

  • low_memory
    (bool, default: False ) –

    Whether to delete data for each node after it is run

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

    The size used for internal batching. Overrides environment variable if set.

  • profile
    (bool, default: False ) –

    whether to log to INFO level the memory usage

set_default 

set_default() -> None

Set the current run as the default for the collection.

Makes it immutable, then moves the default pointer to it.

lookup_key 

lookup_key(from_source: str, to_sources: list[str], key: str) -> dict[str, list[str]]

Matches IDs against the selected backend.

Parameters:

  • from_source
    (str) –

    Name of source the provided key belongs to

  • to_sources
    (list[str]) –

    Names of sources to find keys in

  • key
    (str) –

    The value to match from the source. Usually a primary key

Returns:

  • dict[str, list[str]] –

    Dictionary mapping source names to list of keys within that source.

Examples:

dag.lookup_key(
    from_source="companies_house",
    to_sources=[
        "datahub_companies",
        "hmrc_exporters",
    ]
    key="8534735",
)

get_matches 

get_matches(resolver: ResolverStepName | None = None, source_filter: list[SourceStepName] | None = None, location_names: list[str] | None = None) -> ResolverMatches

Return ResolverMatches, optionally filtering.

Parameters:

  • resolver
    (ResolverStepName | None, default: None ) –

    Name of resolver to query within DAG. If not provided, will look for an apex.

  • source_filter
    (list[SourceStepName] | None, default: None ) –

    An optional list of source step names to filter by.

  • location_names
    (list[str] | None, default: None ) –

    An optional list of location names to filter by.