Sources¶

Generated from the docstrings. Source Connectors shows each connector in use.

sources ¶

Data source connectors for statcatlib.

ingest_ckan ¶

ingest_ckan(
    store: WorkingCopy,
    base_url: str,
    *,
    organization: str | None = None,
    subject: str | None = None,
    limit: int = 100,
) -> list[OpenDataset]

Ingest datasets from a CKAN portal into the catalog.

Fetches up to limit packages from the CKAN Action API, maps each to a statcatlib OpenDataset, stores them all, and returns the list. It also records the language pairs the portal's dataset schema declares (CkanClient.alternate_languages), with base_url as their source.

When the schema lists choices for geographic_region, they are stored as a ConceptScheme (CkanClient.field_choices), and each region code a package carries becomes a spatial coverage whose uri is that code's concept. Its labels are the concept's, in every language the portal sends.

Parameters:

Name	Type	Description	Default
`store`	`WorkingCopy`	The working copy to write into.	required
`base_url`	`str`	CKAN portal base URL, e.g. `"https://open.canada.ca/data"`.	required
`organization`	`str \| None`	Optional CKAN organization name slug filter, e.g. `"statcan-statcan"`.	`None`
`subject`	`str \| None`	Optional free-text subject filter appended to the search query.	`None`
`limit`	`int`	Maximum number of datasets to ingest.	`100`

Returns:

Type	Description
`list[OpenDataset]`	List of OpenDataset objects ingested. A store that is not a
`list[OpenDataset]`	`WorkingCopy`, such as an `EditionView`, raises `TypeError`.

dataset_from_dataflow ¶

dataset_from_dataflow(
    flow: Dataflow,
    dsd: DataStructure,
    *,
    publisher: Agent | None = None,
    themes: list[ConceptRef] | None = None,
) -> AggregateDataset

Create an AggregateDataset from sdmxlib Dataflow and DataStructure objects.

Parameters:

Name	Type	Description	Default
`flow`	`Dataflow`	The sdmxlib Dataflow to convert.	required
`dsd`	`DataStructure`	The resolved DataStructure for this dataflow.	required
`publisher`	`Agent \| None`	Optional Agent override. When omitted, the flow's maintainer agency is used.	`None`
`themes`	`list[ConceptRef] \| None`	Optional list of ConceptRef objects to tag the dataset with.	`None`

Returns:

Type	Description
`AggregateDataset`	The catalog dataset. Code and codelist information is resolved at
`AggregateDataset`	query time via the SDMX structural tables (`sdmx.dsd_component`,
`AggregateDataset`	`sdmx.code`) using the `sdmx_dataflow_urn` foreign key.

ingest_data ¶

ingest_data(
    store: WorkingCopy,
    reg: ObservationSource,
    identifier: str,
    *,
    measure_dim: str | None = None,
) -> str

Materialise observation data for one dataflow into the working copy's lake.

Fetches the observations from reg in the long layout, one row per observation with a column per dimension and measure, and writes them into <agency>.df_<id>_<version> in the edition's lake through wc.registry.write_observations. A re-run replaces the table. It then registers a sdmxlib.DataflowBinding for that table, so the dataflow reads back through the edition's registry::

obs = view.registry.obs(flow_urn)
obs.filter(obs.col("GEO") == "CA").frame()
obs.pivot(on=ds.measure_dim)

Requires that ingest_sdmx has already been called for this identifier, so the catalog record and the DSD are stored.

Parameters:

Name	Type	Description	Default
`store`	`WorkingCopy`	The working copy holding the `AggregateDataset` for identifier.	required
`reg`	`ObservationSource`	Where to fetch observations from, usually a `sdmxlib.RestRegistry`.	required
`identifier`	`str`	Catalog identifier in `AGENCY:ID(VERSION)` form. `(latest)` resolves to the highest stored version.	required
`measure_dim`	`str \| None`	The dimension readers pivot on, recorded as `AggregateDataset.measure_dim`. It does not change what is stored. `None` (default) records none, and `pivot()` then uses the DSD's `MeasureDimension`.	`None`

Returns:

Type	Description
`str`	The fully-qualified table location (`<schema>.<table>`).

A store that is not a WorkingCopy, such as an EditionView, raises TypeError.

Raises:

Type	Description
`KeyError`	No `AggregateDataset` is stored under identifier — run `ingest_sdmx` first.
`ValueError`	The dataset carries no `sdmx_dataflow_urn`, so there is nothing to fetch observations for; identifier is not in `AGENCY:ID(VERSION)` form; the dataflow's DSD is not stored, so no binding can describe the table; or measure_dim is not one of the DSD's dimensions.

ingest_dataflows ¶

ingest_dataflows(
    wc: WorkingCopy,
    source: RestRegistry | EditionRegistry,
    dataflows: Sequence[SdmxUrn[Dataflow] | str],
    *,
    observations: Mapping[str, WriteSource] | None = None,
    partials: bool = True,
    publisher: Agent | None = None,
    themes: list[ConceptRef] | None = None,
) -> list[AggregateDataset]

Fill a data product's working copy with dataflows, cutting their codelists, and store their datasets.

It calls WorkingCopy.fill_dataflows, which copies each dataflow's closure from source, writes its observations, cuts each codelist to the codes the product uses and records what each partial was cut from. Then it builds an AggregateDataset for each dataflow given, read back from the working copy with its data structure, and stores them in one bulk_load. The other dataflows a republished product holds keep their datasets.

Parameters:

Name	Type	Description	Default
`wc`	`WorkingCopy`	The working copy of a data product.	required
`source`	`RestRegistry \| EditionRegistry`	Where the structures are: a `sdmxlib.RestRegistry`, or a sealed edition's or a scratch `sdmxlib.EditionRegistry`.	required
`dataflows`	`Sequence[SdmxUrn[Dataflow] \| str]`	The dataflows, by URN. A `latest` URN resolves to the version the fill stored.	required
`observations`	`Mapping[str, WriteSource] \| None`	The rows of each dataflow that has any, by URN string.	`None`
`partials`	`bool`	Whether to cut codelists to partials.	`True`
`publisher`	`Agent \| None`	The publisher every dataset records; the dataflow's maintenance agency when `None`.	`None`
`themes`	`list[ConceptRef] \| None`	The themes every dataset records.	`None`

Returns:

Type	Description
`list[AggregateDataset]`	The datasets stored, one per dataflow, in the order given.

Raises:

Type	Description
`LookupError`	A dataflow or its data structure cannot be read back from the working copy.

A wc that is not a WorkingCopy, such as an EditionView, raises TypeError, as does a structure product's working copy.

ingest_sdmx ¶

ingest_sdmx(
    store: WorkingCopy, registry: InMemoryRegistry
) -> list[AggregateDataset]

Ingest all SDMX artefacts from a registry into a working copy.

Step 1: Write all structural artefacts (codelists, DSDs, dataflows) into the edition's store through the working copy's sdmxlib registry.

Step 2: For each Dataflow in the registry, create an AggregateDataset via dataset_from_dataflow() and store it. Code and codelist queries use the SDMX structural tables via sdmx_dataflow_urn.

Parameters:

Name	Type	Description	Default
`store`	`WorkingCopy`	The working copy to write into.	required
`registry`	`InMemoryRegistry`	The SDMX artefacts.	required

Returns:

Type	Description
`list[AggregateDataset]`	The AggregateDataset objects created. A store that is not a
`list[AggregateDataset]`	`WorkingCopy`, such as an `EditionView`, raises `TypeError`.

ingest_sdmx_registry ¶

ingest_sdmx_registry(
    store: WorkingCopy,
    registry: InMemoryRegistry,
    *,
    publisher: Agent | None = None,
    themes: list[ConceptRef] | None = None,
) -> list[AggregateDataset]

Bulk ingest all dataflows from an sdmxlib registry into a working copy.

Like ingest_sdmx() but exposes publisher and themes overrides that apply to every ingested dataset.

Parameters:

Name	Type	Description	Default
`store`	`WorkingCopy`	The working copy to write into.	required
`registry`	`InMemoryRegistry`	The SDMX artefacts.	required
`publisher`	`Agent \| None`	The publisher every dataset records.	`None`
`themes`	`list[ConceptRef] \| None`	The themes every dataset records.	`None`

Returns:

Type	Description
`list[AggregateDataset]`	The AggregateDataset objects created. A store that is not a
`list[AggregateDataset]`	`WorkingCopy`, such as an `EditionView`, raises `TypeError`.

prepare_sdmx ¶

prepare_sdmx(
    registry: InMemoryRegistry | LocalRegistry,
    *,
    publisher: Agent | None = None,
    themes: list[ConceptRef] | None = None,
) -> list[AggregateDataset]

Build AggregateDataset objects from a registry without storing.

Callers can enrich each dataset before storing::

datasets = prepare_sdmx(registry)
ds = attrs.evolve(datasets[0], num_series=42)
store.put(ds)

Code and codelist information is resolved at query time via the SDMX structural tables once local.ingest(registry) has been called.

registry can also be a LocalRegistry whose store already holds the structures, such as an edition's: each flow's DSD is read from the store.