API reference — Dataspot source¶

The Dataspot catalog source: extraction, field mapping, vocabulary mapping, enrichment and structure handling. The field and vocabulary mappings are also documented as tables on the Mappings page.

source ¶

drain_extract_skips ¶

drain_extract_skips() -> list[dict]

Return and clear the skips recorded by the last extract.

Each dict matches the invalid_details shape the alert email consumes (id/title/stage/kind/field_errors) plus resource so the caller can attach it to the right raw asset.

Source code in src/metadataswiss_connector/sources/dataspot/source.py

def drain_extract_skips() -> list[dict]:
    """Return and clear the skips recorded by the last extract.

    Each dict matches the ``invalid_details`` shape the alert email
    consumes (``id``/``title``/``stage``/``kind``/``field_errors``) plus
    ``resource`` so the caller can attach it to the right raw asset.
    """
    skips = list(_extract_skips)
    _extract_skips.clear()
    return skips

dataspot_source ¶

dataspot_source(
    base_url: str = dlt.config.value,
    database_name: str = dlt.config.value,
    exposed_client_id: str = dlt.config.value,
    tenant_id: str = dlt.secrets.value,
    client_id: str = dlt.secrets.value,
    client_secret: str = dlt.secrets.value,
    dataspot_access_key: str = dlt.secrets.value,
)

dlt source for the Dataspot metadata catalog REST API.

Arguments not passed explicitly are injected by dlt's config/secrets system from the sources.dataspot.* section — i.e. env vars under the SOURCES__DATASPOT__* naming (loaded from .env; see .env.example). The explicit section pins that lookup path independently of this module's name.

Source code in src/metadataswiss_connector/sources/dataspot/source.py

@dlt.source(name="dataspot", section="dataspot")
def dataspot_source(
    base_url: str = dlt.config.value,
    database_name: str = dlt.config.value,
    exposed_client_id: str = dlt.config.value,
    tenant_id: str = dlt.secrets.value,
    client_id: str = dlt.secrets.value,
    client_secret: str = dlt.secrets.value,
    dataspot_access_key: str = dlt.secrets.value,
):
    """dlt source for the Dataspot metadata catalog REST API.

    Arguments not passed explicitly are injected by dlt's config/secrets
    system from the ``sources.dataspot.*`` section — i.e. env vars under
    the ``SOURCES__DATASPOT__*`` naming (loaded from ``.env``; see
    ``.env.example``). The explicit ``section`` pins that lookup path
    independently of this module's name.
    """
    auth = DataspotAuth(
        access_token_url=f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
        client_id=client_id,
        client_secret=client_secret,
        access_token_request_data={"scope": f"api://{exposed_client_id}/.default"},
        dataspot_access_key=dataspot_access_key,
    )

    api_base_url = f"{base_url}/rest/{database_name}"

    # Broad fetch over every public data product regardless of
    # stereotype. Downstream filter transformers split this into
    # ``data_products`` (OGD/GEO → I14Y Dataset) and
    # ``data_services`` (API → I14Y DataService). Ancestry walks
    # also consume this broad stream so contact points / data
    # owners get resolved for API records too.
    products_spec = _public_resource(
        "data_products_all",
        "schemes/Datenprodukte/datasets",
        "_embedded.datasets",
    )
    # Record id → label so skip entries (see _skip_responses) can carry a
    # human-readable title into the alert email.
    products_spec["processing_steps"].append({"map": _remember_label})

    # Code lists opt in to I14Y via the same ``publish_on_i14y`` custom
    # property as data products (it surfaces on the enumeration list item).
    # ``replace`` — like data_products/data_services — so a code list that
    # loses its ``"yes"`` drops out and is decommissioned; a merge table
    # would keep publishing the stale row. Filtering the parent also stops
    # ``code_list_entries`` (its child) from fetching literals for lists
    # that won't be published.
    code_lists_spec = _public_resource(
        "code_lists",
        "schemes/Referenzdaten/enumerations",
        "_embedded.enumerations",
    )
    code_lists_spec["processing_steps"].append({"filter": _publish_on_i14y})
    code_lists_spec["write_disposition"] = "replace"

    config = {
        "client": {"base_url": api_base_url, "auth": auth},
        "resource_defaults": {"primary_key": "id", "write_disposition": "merge"},
        "resources": [
            products_spec,
            code_lists_spec,
            _public_resource(
                "code_list_entries",
                "schemes/Referenzdaten/enumerations/{resources.code_lists.id}/literals",
                "_embedded.literals",
                parent=True,
                ignore_statuses=(404,),
            ),
            _public_resource(
                "distributions",
                "datasets/{resources.data_products_all.id}/distributions",
                "_embedded.distributions",
                parent=True,
                ignore_statuses=(404,),
            ),
            _public_resource(
                "compositions",
                "datasets/{resources.data_products_all.id}/compositions",
                "_embedded.compositions",
                parent=True,
                # Dataspot server-side bugs on compositions for some
                # datasets: 500 for some, 400 ("BusinessAttribute ...
                # forbidden") when a composition links an attribute the
                # API user may not see. Skip the dataset's structure
                # rather than fail the whole run.
                ignore_statuses=(400, 404, 500),
            ),
        ],
    }

    resources = list(rest_api_resources(config))
    data_products_all = next(r for r in resources if r.name == "data_products_all")
    compositions = next(r for r in resources if r.name == "compositions")

    enrichment = DataspotEnrichment(
        base_url=base_url, database_name=database_name, auth=auth
    )

    def _iter_ancestor_pairs(items):
        """Yield (item, depth, href, collection) for every (dataset, ancestor)."""
        for item in _as_list(items):
            for depth, (href, collection) in enumerate(enrichment.walk_ancestors(item)):
                yield item, depth, href, collection

    @dlt.transformer(
        data_from=data_products_all,
        name="data_products",
        primary_key="id",
        # ``replace`` (not ``merge``) so the ``publish_on_i14y`` gate below
        # actually removes records: a merge table retains rows no longer
        # yielded, so a dataset that loses its ``"yes"`` would keep being
        # published. Replace rebuilds the table as exactly the opted-in set
        # each run, letting sync decommission the rest.
        write_disposition="replace",
    )
    def data_products(items):
        """OGD/GEO data products opted in to I14Y (published as Datasets)."""
        for item in _as_list(items):
            if (
                item.get("stereotype") in DATASET_STEREOTYPES
                and _publish_on_i14y(item)
            ):
                yield item

    @dlt.transformer(
        data_from=data_products_all,
        name="data_services",
        primary_key="id",
        # ``replace`` for the same reason as ``data_products`` above: the
        # gate must be able to drop an API that loses its ``"yes"``.
        write_disposition="replace",
    )
    def data_services(items):
        """API data products opted in to I14Y (published as DataServices)."""
        for item in _as_list(items):
            if (
                item.get("stereotype") in DATA_SERVICE_STEREOTYPES
                and _publish_on_i14y(item)
            ):
                yield item

    @dlt.transformer(
        data_from=data_products_all,
        name="dataservice_serves_datasets",
        primary_key=["dataservice_id", "dataset_id"],
        write_disposition="replace",
    )
    def dataservice_serves_datasets(items):
        """Per (API, produced-dataset) row for I14Y ``servesDatasets``.

        Only API records carry the ``derivedFrom`` link we care about; for
        each we emit one row per SPEZ2-derived dataset. ``replace`` so a
        derivation removed in Dataspot also drops out of the lookup.
        """
        for item in _as_list(items):
            if item.get("stereotype") not in DATA_SERVICE_STEREOTYPES:
                continue
            for dataset_id in enrichment.fetch_derived_dataset_ids(item["id"]):
                yield {"dataservice_id": item["id"], "dataset_id": dataset_id}

    @dlt.transformer(
        data_from=data_products_all,
        name="collections",
        primary_key="id",
        write_disposition="merge",
    )
    def collections(items):
        emitted: set[str] = set()
        for _item, _depth, href, collection in _iter_ancestor_pairs(items):
            if href in emitted:
                continue
            emitted.add(href)
            yield collection

    @dlt.transformer(
        data_from=data_products_all,
        name="dataset_collection_path",
        primary_key=["dataset_id", "depth"],
        write_disposition="merge",
    )
    def dataset_collection_path(items):
        for item, depth, _href, collection in _iter_ancestor_pairs(items):
            yield {
                "dataset_id": item.get("id"),
                "collection_id": collection.get("id"),
                "depth": depth,
            }

    @dlt.transformer(
        data_from=collections,
        name="collection_agencies",
        primary_key="collection_id",
        write_disposition="merge",
    )
    def collection_agencies(collection_items):
        for collection in _as_list(collection_items):
            if collection.get("stereotype") != STEREOTYPE_ORGANIZATIONAL_UNIT:
                continue
            state_calendar_id = collection.get("customProperties", {}).get(
                "stateCalendarId"
            )
            if state_calendar_id is None:
                continue
            agency = enrichment.fetch_agency(int(state_calendar_id))
            if agency is None:
                continue
            yield {
                "collection_id": collection.get("id"),
                **_agency_lookup_row(int(state_calendar_id), agency),
            }

    @dlt.transformer(
        data_from=data_products,
        name="kontaktstelle_agencies",
        primary_key="state_calendar_id",
        write_disposition="merge",
    )
    def kontaktstelle_agencies(items):
        """Staatskalender agencies referenced by ``i14y_kontaktstelle_sk_id``.

        One row per distinct Staatskalender ID that datasets override their
        Kontaktstelle with. ``fetch_agency`` memoizes per ID for the life
        of the extract, so a handful of IDs shared across many datasets
        cost one Staatskalender call each, no matter how many datasets
        reference them.
        """
        seen: set[int] = set()
        for item in _as_list(items):
            sk_id = item.get("customProperties", {}).get(
                "i14y_kontaktstelle_sk_id"
            )
            if sk_id is None:
                continue
            sk_id = int(sk_id)
            if sk_id in seen:
                continue
            seen.add(sk_id)
            agency = enrichment.fetch_agency(sk_id)
            if agency is None:
                continue
            yield _agency_lookup_row(sk_id, agency)

    @dlt.transformer(
        data_from=data_products_all,
        name="collection_data_owners",
        primary_key="collection_id",
        write_disposition="merge",
    )
    def collection_data_owners(items):
        # Walks ancestors directly from data_products_all rather than from
        # ``collections``: dlt pipes are single-consumer, and
        # ``collection_agencies`` already drains that one.
        emitted: set[str] = set()
        for _item, _depth, _href, collection in _iter_ancestor_pairs(items):
            collection_id = collection.get("id")
            if not collection_id or collection_id in emitted:
                continue
            emitted.add(collection_id)
            attr_href = (
                collection.get("_links", {}).get("attributedTo", {}).get("href")
            )
            if not attr_href:
                continue
            name = enrichment.fetch_data_owner_name(attr_href)
            if not name:
                continue
            yield {"collection_id": collection_id, "name": name}

    @dlt.transformer(
        data_from=compositions,
        name="dataset_structure_components",
        primary_key="composition_id",
        write_disposition="replace",
    )
    def dataset_structure_components(items):
        """Per (dataset, attribute) row with everything SHACL needs.

        Joins composition → attribute → datatype eagerly so the dataset
        transform can build a Turtle SHACL document without further API
        calls. ``replace`` semantics: a structure that loses an attribute
        in Dataspot must lose it on the I14Y side too.
        """
        for composition in _as_list(items):
            attr_href = (
                composition.get("_links", {}).get("composedOf", {}).get("href")
            )
            if not attr_href:
                continue
            attribute = enrichment.fetch_attribute(attr_href)
            if not _is_public(attribute):
                continue
            # Some compositions point back at a Dataset rather than at a
            # real attribute (Dataspot GEO stereotype quirk). Those would
            # produce nonsensical sh:property rows, so drop them.
            if attribute.get("_type") not in {"UmlAttribute", "BusinessAttribute"}:
                continue
            dt_href = (
                attribute.get("_links", {}).get("hasRange", {}).get("href")
            )
            datatype = enrichment.fetch_datatype(dt_href) if dt_href else {}
            if datatype and not _is_public(datatype):
                datatype = {}
            # Enumerations sit at /enumerations/{id} and have
            # _type=ReferenceObject; datatypes sit at /datatypes/{id} with
            # _type=DataDomain. We surface the kind so the SHACL builder
            # can emit dcterms:conformsTo for code-list-typed attributes.
            datatype_kind = (datatype or {}).get("_type")
            yield {
                "dataset_id": composition.get("componentOf"),
                "composition_id": composition.get("id"),
                "order": composition.get("order"),
                "composition_label": composition.get("label"),
                "composition_title": composition.get("title"),
                "composition_description": composition.get("description"),
                "composition_required": (
                    composition.get("customProperties", {}).get("required")
                ),
                "attribute_id": attribute.get("id"),
                "attribute_label": attribute.get("label"),
                "attribute_description": attribute.get("description"),
                "attribute_required": attribute.get("required"),
                "attribute_cardinality": attribute.get("cardinality"),
                "attribute_min_inclusive": attribute.get("minInclusive"),
                "attribute_max_inclusive": attribute.get("maxInclusive"),
                "datatype_id": (datatype or {}).get("id"),
                "datatype_kind": datatype_kind,
                "datatype_base_type": (datatype or {}).get("baseType"),
                "datatype_label": (datatype or {}).get("label"),
            }

    return [
        *resources,
        data_products,
        data_services,
        dataservice_serves_datasets,
        collections,
        dataset_collection_path,
        collection_agencies,
        kontaktstelle_agencies,
        collection_data_owners,
        dataset_structure_components,
    ]

transform ¶

Dataspot-specific field mapping to I14Y DCAT format.

transform_to_dataset ¶

transform_to_dataset(
    record: dict,
    children: dict[str, list],
    *,
    lookups: Lookups,
    publisher: str
) -> (
    tuple[DcatDatasetInputModel, dict]
    | DcatDatasetInputModel
)

Map a flattened Dataspot dataset row to a DcatDatasetInputModel.

Parameters:

Name	Type	Description	Default
`record`	`dict`	Flattened row from DuckDB. Nested fields use dlt's `__` separator (e.g. `custom_properties__publisher`).	required
`children`	`dict[str, list]`	dlt 1:n child tables keyed by field name (e.g. `tags`).	required
`lookups`	`Lookups`	per-run view over the cross-reference tables (e.g. `collections`, `dataset_collection_path`, `collection_agencies`, `collection_data_owners`).	required
`publisher`	`str`	I14Y publisher identifier (e.g. "Basel-Stadt"). Configured globally via I14Y_PUBLISHER_IDENTIFIER env var.	required

Source code in src/metadataswiss_connector/sources/dataspot/transform.py

def transform_to_dataset(
    record: dict,
    children: dict[str, list],
    *,
    lookups: Lookups,
    publisher: str,
) -> tuple[DcatDatasetInputModel, dict] | DcatDatasetInputModel:
    """Map a flattened Dataspot dataset row to a DcatDatasetInputModel.

    Args:
        record: Flattened row from DuckDB. Nested fields use dlt's ``__``
                separator (e.g. ``custom_properties__publisher``).
        children: dlt 1:n child tables keyed by field name (e.g. ``tags``).
        lookups: per-run view over the cross-reference tables (e.g.
                 ``collections``, ``dataset_collection_path``,
                 ``collection_agencies``, ``collection_data_owners``).
        publisher: I14Y publisher identifier (e.g. "Basel-Stadt").
                   Configured globally via I14Y_PUBLISHER_IDENTIFIER env var.
    """
    tags = children.get("tags", [])
    # Sort by the stable distribution id: ``distributions`` is loaded from a
    # sibling table whose SELECT has no inherent order, so without this the
    # list order — and thus the payload hash — can vary between runs.
    distributions = [
        _map_distribution(d)
        for d in sorted(
            children.get("distributions", []),
            key=lambda d: str(d.get("id") or ""),
        )
        if d.get("access_url")
    ]
    override = _contact_points_from_kontaktstelle_override(record, lookups)
    contact_points = (
        override
        if override is not None
        else _organizational_unit_contact_points(record["id"], lookups)
    )
    data_owner = _resolve_data_owner(record["id"], lookups)
    structure_ttl = build_dataset_shacl_turtle(
        dataset_id=record["id"],
        dataset_label=record.get("label"),
        components=lookups.by("dataset_structure_components", "dataset_id").get(
            record["id"], []
        ),
    )
    model = DcatDatasetInputModel(
        data_owner=data_owner,
        title=dcat.multi_language(record.get("label")),
        description=dcat.multi_language(dcat.html_to_plain_text(record.get("description"))),
        identifiers=[record["id"]],
        publisher=IdentifierInputModel(identifier=publisher),
        responsible_person=EmailInputModel(email=RESPONSIBLE_PERSON_EMAIL),
        responsible_deputy=EmailInputModel(email=RESPONSIBLE_DEPUTY_EMAIL),
        access_rights=CodeInputModel(code="PUBLIC"),
        confidentiality_person=mappings.confidentiality_person(
            children.get("custom_properties__personal_data", [])
        ),
        issued=dcat.epoch_ms_to_datetime(
            record.get("custom_properties__publication_date")
        ),
        modified=dcat.epoch_ms_to_datetime(record.get("custom_properties__last_update")),
        frequency=dcat.frequency(record.get("accrual_periodicity")),
        keywords=dcat.keywords(tags),
        themes=mappings.themes(
            children.get("custom_properties__topics", []),
            record.get("custom_properties__geo_category"),
        ),
        spatial=[record["spatial"]] if record.get("spatial") else None,
        temporal_coverage=dcat.temporal_coverage(record.get("temporal_start"), record.get("temporal_end")),
        contact_points=contact_points or None,
        languages=[CodeInputModel(code="de")],
        retention_period_complement=mappings.retention_period_complement(
            record.get("custom_properties__retention_period"),
            record.get("custom_properties__retention_justification"),
        ),
        landing_pages=dcat.landing_pages(record.get("custom_properties__i14y_dataset_landing_page")) or None,
        version=record.get("custom_properties__i14y_dataset_version"),
        version_notes=dcat.multi_language(record.get("custom_properties__i14y_dataset_version_notes")),
        distributions=distributions or None,
    )
    if structure_ttl:
        return model, {"structure": structure_ttl}
    return model

transform_to_dataservice ¶

transform_to_dataservice(
    record: dict,
    children: dict[str, list],
    *,
    lookups: Lookups,
    publisher: str
) -> DataServiceInputModel

Map a flattened Dataspot API record to a DataServiceInputModel.

Dataspot exposes APIs as data products with stereotype API. They share most metadata with OGD/GEO datasets but publish to I14Y under /dataservices with a different input model (no distributions, no SHACL structure). Endpoint metadata comes from dedicated i14y_api_* custom properties on the dataspot record.

Source code in src/metadataswiss_connector/sources/dataspot/transform.py

def transform_to_dataservice(
    record: dict,
    children: dict[str, list],
    *,
    lookups: Lookups,
    publisher: str,
) -> DataServiceInputModel:
    """Map a flattened Dataspot API record to a DataServiceInputModel.

    Dataspot exposes APIs as data products with stereotype ``API``. They
    share most metadata with OGD/GEO datasets but publish to I14Y under
    ``/dataservices`` with a different input model (no distributions, no
    SHACL structure). Endpoint metadata comes from dedicated
    ``i14y_api_*`` custom properties on the dataspot record.
    """
    tags = children.get("tags", [])
    contact_points = _organizational_unit_contact_points(record["id"], lookups)
    return DataServiceInputModel(
        title=dcat.multi_language(record.get("label")),
        description=dcat.multi_language(
            dcat.html_to_plain_text(record.get("description"))
        ),
        identifiers=[record["id"]],
        publisher=IdentifierInputModel(identifier=publisher),
        responsible_person=EmailInputModel(email=RESPONSIBLE_PERSON_EMAIL),
        responsible_deputy=EmailInputModel(email=RESPONSIBLE_DEPUTY_EMAIL),
        serves_datasets=_resolve_serves_datasets(record["id"], lookups) or None,
        access_rights=_dataservice_access_rights(record),
        issued=dcat.epoch_ms_to_datetime(
            record.get("custom_properties__publication_date")
        ),
        modified=dcat.epoch_ms_to_datetime(record.get("custom_properties__last_update")),
        keywords=dcat.keywords(tags),
        contact_points=contact_points or None,
        landing_pages=dcat.landing_pages(
            record.get("custom_properties__i14y_api_landing_page")
        )
        or None,
        endpoint_urls=_dataservice_resource_list(
            record, children, "custom_properties__i14y_api_endpoint_url"
        )
        or None,
        endpoint_descriptions=_dataservice_resource_list(
            record, children, "custom_properties__i14y_api_endpoint_description"
        )
        or None,
    )

transform_to_concept ¶

transform_to_concept(
    record: dict,
    children: dict[str, list],
    *,
    lookups: Lookups,
    publisher: str
) -> tuple[CodeListConceptInput, dict]

Map a Dataspot enumeration row to a CodeListConceptInput.

Returns (model, extras) where extras["entries"] carries the code-list entry payloads. The concept itself is created via POST /concepts; the entries are uploaded as a follow-up via POST /concepts/{id}/codelist-entries/imports/Json once the concept's I14Y UUID is known.

Source code in src/metadataswiss_connector/sources/dataspot/transform.py

def transform_to_concept(
    record: dict,
    children: dict[str, list],
    *,
    lookups: Lookups,
    publisher: str,
) -> tuple[CodeListConceptInput, dict]:
    """Map a Dataspot enumeration row to a CodeListConceptInput.

    Returns ``(model, extras)`` where ``extras["entries"]`` carries the
    code-list entry payloads. The concept itself is created via
    ``POST /concepts``; the entries are uploaded as a follow-up via
    ``POST /concepts/{id}/codelist-entries/imports/Json`` once the
    concept's I14Y UUID is known.
    """
    # Enumerations carry no timestamp custom property — ``date_created`` is
    # the only timestamp available in the source payload. Fall back to the
    # fixed open-lower-bound sentinel rather than ``datetime.now()``: a
    # wall-clock default would change the payload hash on every run and
    # trigger a spurious re-publish for any concept missing date_created.
    valid_from = (
        dcat.epoch_ms_to_datetime(record.get("date_created"))
        or dcat.epoch_ms_to_datetime(DATASPOT_VALID_FROM_SENTINEL)
    )
    # Sort by a stable entry key: ``code_list_entries`` is a dlt child
    # table whose read order isn't guaranteed, so without this the entries
    # list order — and thus the payload hash — can vary between runs.
    raw_entries = sorted(
        children.get("code_list_entries", []),
        key=lambda e: (str(e.get("code") or ""), str(e.get("id") or "")),
    )
    model = CodeListConceptInput(
        identifiers=[record["id"]],
        name=dcat.multi_language(record.get("label")),
        description=(
            dcat.multi_language(dcat.html_to_plain_text(record.get("description")))
            or dcat.multi_language(record.get("label"))
        ),
        publisher=IdentifierInputModel(identifier=publisher),
        responsible_person=EmailInputModel(email=RESPONSIBLE_PERSON_EMAIL),
        responsible_deputy=EmailInputModel(email=RESPONSIBLE_DEPUTY_EMAIL),
        valid_from=valid_from,
        version=CONCEPT_VERSION,
        code_list_entry_value_type=_value_type_from_entries(raw_entries),
        code_list_entry_value_max_length=_max_code_length(raw_entries),
        code_list_entry_default_sort_property=CodeListEntrySortProperty.code,
    )
    extras = {"entries": _build_entries(raw_entries)}
    return model, extras

mappings ¶

Dataspot-specific vocabulary mappings to I14Y codes.

Kept separate from the generic I14Y builders so that vocabulary changes on the Dataspot side don't ripple into shared code, and other sources don't inherit Dataspot's input shapes.

themes ¶

themes(
    topics: list[str] | None, geo_category: str | None
) -> list[CodeInputModel] | None

Map Dataspot topics (BFS) and geoCategory (eCH-166) to I14Y theme codes.

Unknown values are dropped; duplicates are collapsed; insertion order is preserved. Returns None when nothing maps.

Source code in src/metadataswiss_connector/sources/dataspot/mappings.py

def themes(
    topics: list[str] | None,
    geo_category: str | None,
) -> list[CodeInputModel] | None:
    """Map Dataspot topics (BFS) and geoCategory (eCH-166) to I14Y theme codes.

    Unknown values are dropped; duplicates are collapsed; insertion order
    is preserved. Returns ``None`` when nothing maps.
    """
    codes: list[str] = []
    seen: set[str] = set()
    for topic in topics or []:
        code = _TOPIC_TO_THEME.get(topic)
        if code and code not in seen:
            seen.add(code)
            codes.append(code)
    if geo_category:
        code = _GEOCATEGORY_TO_THEME.get(geo_category)
        if code and code not in seen:
            seen.add(code)
            codes.append(code)
    if not codes:
        return None
    return [CodeInputModel(code=c) for c in codes]

confidentiality_person ¶

confidentiality_person(
    values: list[str] | None,
) -> CodeInputModel | None

Map Dataspot personal_data values to an I14Y confidentialityPerson code.

Source code in src/metadataswiss_connector/sources/dataspot/mappings.py

def confidentiality_person(values: list[str] | None) -> CodeInputModel | None:
    """Map Dataspot ``personal_data`` values to an I14Y confidentialityPerson code."""
    if not values:
        return None
    present = set(values)
    for value, code in _PERSONAL_DATA_PRIORITY:
        if value in present:
            return CodeInputModel(code=code)
    return None

retention_period_complement ¶

retention_period_complement(
    period_years: int | str | None,
    justification: str | None,
    lang: str = "de",
) -> MultiLanguageModel | None

Combine Dataspot's retention period + justification into a plain-text block.

I14Y has no structured field for retention metadata, so the two Dataspot custom properties are concatenated into a single description.

Source code in src/metadataswiss_connector/sources/dataspot/mappings.py

def retention_period_complement(
    period_years: int | str | None,
    justification: str | None,
    lang: str = "de",
) -> MultiLanguageModel | None:
    """Combine Dataspot's retention period + justification into a plain-text block.

    I14Y has no structured field for retention metadata, so the two
    Dataspot custom properties are concatenated into a single description.
    """
    parts = []
    if period_years not in (None, ""):
        parts.append(f"Aufbewahrungsfrist (Jahre): {period_years}")
    if justification:
        parts.append(f"Begründung: {justification}")
    if not parts:
        return None
    return MultiLanguageModel(**{lang: "\n".join(parts)})

enrichment ¶

Enrichment helpers for the Dataspot source.

Wraps the Dataspot REST client and the Staatskalender lookup behind a single class so the source-level dlt transformers stay declarative. Caches and retry logic live here so they're independently testable.

DataspotEnrichment ¶

Resolves ancestry, data-owner names, and staatskalender agencies.

All fetch methods are memoized for the lifetime of the instance. Pass an existing client (and optional staatskalender_client) to inject test doubles; otherwise both are constructed from base_url/database_name/auth.