Skip to content

API reference — DCAT layer

Source-neutral building of I14Y DCAT models, the transform step over DuckDB, schema introspection, cross-reference lookups and the SHACL/Turtle builder.

Note

dcat/i14y_models.py is intentionally omitted here: it is generated from the I14Y OpenAPI specification (see the README) and is not meant to be read as hand-written reference.

builders

I14Y DCAT Input-Model builders.

Helpers that construct typed I14Y *InputModel instances. Each helper takes the loose, source-side primitives (a string, an epoch timestamp, a list of tags) and returns a Pydantic model that matches the I14Y Partner API contract exactly. Validation happens here, not at POST time.

The model classes themselves are generated from the I14Y OpenAPI spec into i14y_models.py — see that file's header for the regen command.

This module is source-agnostic: it knows the I14Y data model and the EU-level controlled vocabularies, nothing about any specific catalog source. Source-specific vocabulary mappings live next to the source (e.g. sources/dataspot/mappings.py).

html_to_plain_text 

html_to_plain_text(value: str | None) -> str | None

Strip HTML markup so the value can be sent as plain text.

Source descriptions may contain inline HTML (<p>…</p>, <br>, entity references). I14Y stores descriptions as unformatted plain text, so all tags are removed and whitespace is collapsed.

Source code in src/metadataswiss_connector/dcat/builders.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def html_to_plain_text(value: str | None) -> str | None:
    """Strip HTML markup so the value can be sent as plain text.

    Source descriptions may contain inline HTML (``<p>…</p>``, ``<br>``,
    entity references). I14Y stores descriptions as unformatted plain
    text, so all tags are removed and whitespace is collapsed.
    """
    if not value:
        return value
    if "<" not in value and "&" not in value:
        return value
    parser = _HTMLStripper()
    parser.feed(value)
    parser.close()
    text = unescape(parser.get_text())
    text = re.sub(r"\s+", " ", text).strip()
    return text or None

file_format 

file_format(value: str | None) -> CodeInputModel | None

Map a loose file-format string to an I14Y file-type vocabulary code.

I14Y uses the EU file-type authority codes (uppercase). Returns None for unknown or missing values so the field is omitted.

Source code in src/metadataswiss_connector/dcat/builders.py 
102
103
104
105
106
107
108
109
110
111
112
113
def file_format(value: str | None) -> CodeInputModel | None:
    """Map a loose file-format string to an I14Y file-type vocabulary code.

    I14Y uses the EU file-type authority codes (uppercase).
    Returns ``None`` for unknown or missing values so the field is omitted.
    """
    if not value:
        return None
    code = _FILE_FORMAT_CODES.get(value.strip().lower())
    if not code:
        return None
    return CodeInputModel(code=code)

frequency 

frequency(uri: str | None) -> CodeInputModel | None

Build a CodeInputModel from an EU frequency URI.

Maps to the I14Y-supported VOCAB_EU_FREQUENCY codes; falls back to OTHER when the value is outside the controlled vocabulary.

Source code in src/metadataswiss_connector/dcat/builders.py 
125
126
127
128
129
130
131
132
133
134
135
136
def frequency(uri: str | None) -> CodeInputModel | None:
    """Build a CodeInputModel from an EU frequency URI.

    Maps to the I14Y-supported VOCAB_EU_FREQUENCY codes; falls back to
    ``OTHER`` when the value is outside the controlled vocabulary.
    """
    if not uri:
        return None
    code = uri.rsplit("/", 1)[-1].upper() if "/" in uri else uri.upper()
    if code not in _FREQUENCY_CODES:
        code = "OTHER"
    return CodeInputModel(code=code)

contact_points 

contact_points(email: str | None) -> list[VCardModel]

Build a VCardModel list from an email address.

Source code in src/metadataswiss_connector/dcat/builders.py 
154
155
156
157
158
def contact_points(email: str | None) -> list[VCardModel]:
    """Build a VCardModel list from an email address."""
    if not email:
        return []
    return [VCardModel(has_email=email)]

epoch_ms_to_datetime 

epoch_ms_to_datetime(
    epoch_ms: int | float | None,
) -> datetime | None

Convert epoch milliseconds to a timezone-aware datetime.

Source code in src/metadataswiss_connector/dcat/builders.py 
167
168
169
170
171
def epoch_ms_to_datetime(epoch_ms: int | float | None) -> datetime | None:
    """Convert epoch milliseconds to a timezone-aware datetime."""
    if epoch_ms is None:
        return None
    return datetime.fromtimestamp(epoch_ms / 1000, tz=timezone.utc)

transforms

Generic raw-to-DCAT transform pipeline step.

Reads raw data from DuckDB (loaded by a source-specific dlt pipeline), applies a source-specific transform function, and loads the result into the i14y_dcat dataset.

run_transform 

run_transform(
    pipeline_raw: Pipeline,
    resources: "dict[str, ResourceSpec]",
    destination,
    *,
    publisher: str,
    extra_lookups: dict[str, list[dict]] | None = None
) -> None

Transform raw resources to I14Y input models and load into DuckDB.

resources maps each raw resource/table name to its ResourceSpec (transform + target kind), allowing one source to carry multiple resource types.

Source code in src/metadataswiss_connector/dcat/transforms.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def run_transform(
    pipeline_raw: dlt.Pipeline,
    resources: "dict[str, ResourceSpec]",
    destination,
    *,
    publisher: str,
    extra_lookups: dict[str, list[dict]] | None = None,
) -> None:
    """Transform raw resources to I14Y input models and load into DuckDB.

    ``resources`` maps each raw resource/table name to its ResourceSpec
    (transform + target kind), allowing one source to carry multiple
    resource types.
    """
    for resource_name, spec in resources.items():
        run_transform_resource(
            pipeline_raw=pipeline_raw,
            resource_name=resource_name,
            spec=spec,
            destination=destination,
            publisher=publisher,
            extra_lookups=extra_lookups,
        )

run_transform_resource 

run_transform_resource(
    pipeline_raw: Pipeline,
    resource_name: str,
    spec: "ResourceSpec",
    destination,
    *,
    publisher: str,
    extra_lookups: dict[str, list[dict]] | None = None
) -> dict

Transform a single raw resource and load into the i14y_dcat dataset.

extra_lookups are synthetic lookup tables merged over the raw sibling tables before they reach the transform — used by the pipeline to expose e.g. the published-ID maps from sync state.

Returns a stats dict {valid, invalid, raw_rows, loaded_tables} so callers (e.g. Dagster assets) can surface it as materialization metadata and run validation-rate checks against it.

Source code in src/metadataswiss_connector/dcat/transforms.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def run_transform_resource(
    pipeline_raw: dlt.Pipeline,
    resource_name: str,
    spec: "ResourceSpec",
    destination,
    *,
    publisher: str,
    extra_lookups: dict[str, list[dict]] | None = None,
) -> dict:
    """Transform a single raw resource and load into the i14y_dcat dataset.

    ``extra_lookups`` are synthetic lookup tables merged over the raw
    sibling tables before they reach the transform — used by the
    pipeline to expose e.g. the published-ID maps from sync state.

    Returns a stats dict ``{valid, invalid, raw_rows, loaded_tables}`` so
    callers (e.g. Dagster assets) can surface it as materialization
    metadata and run validation-rate checks against it.
    """
    pipeline_dcat = dlt.pipeline(
        pipeline_name=f"{pipeline_raw.pipeline_name}_dcat",
        destination=destination,
        dataset_name="i14y_dcat",
    )
    stats: dict = {"valid": 0, "invalid": 0, "invalid_details": []}
    load_info = pipeline_dcat.run(
        _read_and_transform(
            pipeline_raw, resource_name, spec.transform,
            publisher=publisher, stats=stats, extra_lookups=extra_lookups,
            sibling_parent=spec.sibling_parent,
        ),
        table_name=resource_name,
        write_disposition="replace",
    )
    logger.info("%s", load_info)
    row_counts = pipeline_dcat.last_trace.last_normalize_info.row_counts
    loaded_tables = {
        t: c for t, c in row_counts.items() if not t.startswith("_dlt")
    }
    return {
        "valid": stats["valid"],
        "invalid": stats["invalid"],
        "raw_rows": stats["valid"] + stats["invalid"],
        "loaded_tables": loaded_tables,
        "invalid_details": stats["invalid_details"],
    }

duckdb_io

DuckDB read-path for transformed DCAT records.

Re-nests dlt's parent__child flat columns back into nested dicts and re-attaches child tables so the result matches the I14Y camelCase input contract and can be fed directly into DcatDatasetInputModel.model_validate(). The dlt-convention plumbing (child-table discovery, _dlt_list_idx ordering, unflattening) is shared with the transform read path via dlt_schema.

Also owns the two constants describing how post-sync "extras" sidecar payloads are carried through DuckDB (column name on write, key on read).

read_transformed 

read_transformed(
    table_name: str, *, limit: int | None = None
) -> list[dict]

Read transformed DCAT records from DuckDB and reconstruct nested dicts.

dlt flattens nested models into parent__child columns. This function re-nests them so the result matches the I14Y camelCase input contract and can be fed directly into DcatDatasetInputModel.model_validate().

Source code in src/metadataswiss_connector/dcat/duckdb_io.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def read_transformed(
    table_name: str, *, limit: int | None = None
) -> list[dict]:
    """Read transformed DCAT records from DuckDB and reconstruct nested dicts.

    dlt flattens nested models into ``parent__child`` columns.  This function
    re-nests them so the result matches the I14Y camelCase input contract and
    can be fed directly into ``DcatDatasetInputModel.model_validate()``.
    """
    con = duckdb.connect(duckdb_path(), read_only=True)
    try:
        sql = f'SELECT * FROM i14y_dcat."{table_name}"'
        if limit:
            sql += f" LIMIT {limit}"
        try:
            rows = con.execute(sql).fetchall()
        except duckdb.CatalogException:
            # dlt's write_disposition="replace" infers the table schema from
            # the rows it receives, so when every source record of a resource
            # is dropped at validation (e.g. a mandatory field like
            # ``description`` is missing on the Dataspot record) the
            # i14y_dcat table is never created. Surface that here as a clear,
            # actionable message instead of letting the raw "table does not
            # exist" error mask the cause. The specific record id and field
            # are logged by the transform step (see transforms.py) and
            # surfaced as the transformed asset's invalid_records metadata.
            logger.warning(
                "No transformed table i14y_dcat.%s — the transform produced "
                "zero valid records. Every source record was dropped at "
                "validation, typically because a mandatory field is missing; "
                "see the transform step's invalid-records log for the "
                "offending id(s) and field(s). Nothing to publish.",
                table_name,
            )
            return []
        columns = [col[0] for col in con.description]

        # Load child tables (one-to-many fields like keywords, identifiers)
        child_tables = load_child_tables(duckdb_query(con), "i14y_dcat", table_name)

        results = []
        for row in rows:
            flat = dict(zip(columns, row))
            dlt_id = flat.pop("_dlt_id", None)
            flat.pop("_dlt_load_id", None)
            extras_raw = flat.pop(EXTRAS_COLUMN, None)
            record = unflatten(flat)
            # Attach child table data
            for child_name, child_rows in child_tables.items():
                if dlt_id in child_rows:
                    record[child_name] = child_rows[dlt_id]
            if extras_raw:
                try:
                    record[EXTRAS_KEY] = json.loads(extras_raw)
                except (ValueError, TypeError):
                    logger.warning(
                        "Failed to decode extras column on %s id=%r",
                        table_name, record.get("identifiers") or record.get("id"),
                    )
            results.append(record)
        return results
    finally:
        con.close()

dlt_schema

dlt-schema introspection and re-nesting over DuckDB.

Single home for the dlt naming conventions: child tables named <parent>__<field>, _dlt_* system columns, list order via _dlt_list_idx, and a__b__c column flattening. Both read paths — the transform step (dlt sql_client) and the publish step (raw duckdb connection) — share the loaders here through a small query adapter, so rules like "ORDER BY _dlt_list_idx or the payload hash flips" live in exactly one place.

Loaded shapes:

  • scalar children: child tables with a single value data column (dlt's shape for list[str] fields like keywords), yielded as bare values,
  • struct children: child tables with real columns, yielded as re-nested dicts,
  • sibling tables: top-level resources in the same dataset, either wired to a parent via include_from_parent (carrying a _<parent>_<key> ref column) or unanchored lookup tables.

Source-agnostic — only knows the dlt naming conventions.

duckdb_query 

duckdb_query(con: DuckDBPyConnection) -> QueryFn

Adapt a raw duckdb connection to QueryFn.

Source code in src/metadataswiss_connector/dcat/dlt_schema.py 
52
53
54
55
56
57
58
59
def duckdb_query(con: duckdb.DuckDBPyConnection) -> QueryFn:
    """Adapt a raw duckdb connection to ``QueryFn``."""

    def query(sql: str) -> tuple[list[str], list[tuple]]:
        rows = con.execute(sql).fetchall()
        return [col[0] for col in con.description], rows

    return query

sql_client_query 

sql_client_query(client) -> QueryFn

Adapt a dlt sql_client to QueryFn.

Source code in src/metadataswiss_connector/dcat/dlt_schema.py 
62
63
64
65
66
67
68
69
def sql_client_query(client) -> QueryFn:
    """Adapt a dlt ``sql_client`` to ``QueryFn``."""

    def query(sql: str) -> tuple[list[str], list[tuple]]:
        with client.execute_query(sql) as cursor:
            return [col[0] for col in cursor.description], cursor.fetchall()

    return query

unflatten 

unflatten(flat: dict) -> dict

Convert dlt's a__b__c flat keys back into nested dicts.

Example: {"title__de": "Foo"} → {"title": {"de": "Foo"}}

Drops keys whose values are None.

Source code in src/metadataswiss_connector/dcat/dlt_schema.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def unflatten(flat: dict) -> dict:
    """Convert dlt's ``a__b__c`` flat keys back into nested dicts.

    Example: ``{"title__de": "Foo"}`` → ``{"title": {"de": "Foo"}}``

    Drops keys whose values are None.
    """
    nested: dict = {}
    for key, value in flat.items():
        if value is None:
            continue
        parts = key.split("__")
        target = nested
        for part in parts[:-1]:
            target = target.setdefault(part, {})
        target[parts[-1]] = value
    return nested

load_child_tables 

load_child_tables(
    query: QueryFn,
    schema: str,
    parent_table: str,
    *,
    scalars_only: bool = False
) -> dict[str, dict[str, list]]

Discover and load dlt child tables grouped by parent _dlt_id.

Returns {child_field: {parent_dlt_id: [item, ...]}}. Scalar-list children yield bare values; struct children yield re-nested dicts (or are skipped entirely with scalars_only). Rows are ordered by _dlt_list_idx so list elements keep their source order deterministically across runs — without it DuckDB's physical row order can vary, perturbing the payload hash and triggering spurious re-publishes.

Source code in src/metadataswiss_connector/dcat/dlt_schema.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def load_child_tables(
    query: QueryFn,
    schema: str,
    parent_table: str,
    *,
    scalars_only: bool = False,
) -> dict[str, dict[str, list]]:
    """Discover and load dlt child tables grouped by parent ``_dlt_id``.

    Returns ``{child_field: {parent_dlt_id: [item, ...]}}``. Scalar-list
    children yield bare values; struct children yield re-nested dicts
    (or are skipped entirely with ``scalars_only``). Rows are ordered by
    ``_dlt_list_idx`` so list elements keep their source order
    deterministically across runs — without it DuckDB's physical row
    order can vary, perturbing the payload hash and triggering spurious
    re-publishes.
    """
    prefix = f"{parent_table}__"
    result: dict[str, dict[str, list]] = {}
    for table in _tables_in_schema(query, schema):
        if not table.startswith(prefix):
            continue
        field_name = table[len(prefix):]
        try:
            columns, rows = query(
                f'SELECT * FROM "{schema}"."{table}" '
                "ORDER BY _dlt_parent_id, _dlt_list_idx"
            )
        except _UNDEFINED_TABLE_ERRORS:
            continue
        data_columns = [c for c in columns if c not in _DLT_COLUMNS]
        is_scalar_list = data_columns == ["value"]
        if scalars_only and not is_scalar_list:
            continue
        by_parent: dict[str, list] = defaultdict(list)
        for row in rows:
            rec = dict(zip(columns, row))
            parent_id = rec.get("_dlt_parent_id")
            if not parent_id:
                continue
            if is_scalar_list:
                by_parent[parent_id].append(rec["value"])
            else:
                by_parent[parent_id].append(unflatten(_strip_dlt_columns(rec)))
        result[field_name] = dict(by_parent)
    return result

load_scalar_children 

load_scalar_children(
    client, dataset_name: str, parent_table: str
) -> dict[str, dict[str, list[str]]]

Load dlt scalar-list child tables grouped by parent, via sql_client.

Returns {child_field: {parent_dlt_id: [values]}}. Only child tables with a single value column (dlt's shape for list[str] fields) are included — struct children are skipped here.

Source code in src/metadataswiss_connector/dcat/dlt_schema.py 
159
160
161
162
163
164
165
166
167
168
169
170
def load_scalar_children(
    client, dataset_name: str, parent_table: str
) -> dict[str, dict[str, list[str]]]:
    """Load dlt scalar-list child tables grouped by parent, via sql_client.

    Returns ``{child_field: {parent_dlt_id: [values]}}``. Only child
    tables with a single ``value`` column (dlt's shape for ``list[str]``
    fields) are included — struct children are skipped here.
    """
    return load_child_tables(
        sql_client_query(client), dataset_name, parent_table, scalars_only=True
    )

load_sibling_children 

load_sibling_children(
    client,
    dataset_name: str,
    parent_table: str,
    *,
    parent_key: str,
    ref_parent: str | None = None
) -> tuple[
    dict[str, dict[str, list[dict]]], dict[str, list[dict]]
]

Load top-level sibling tables alongside the parent.

Returns (siblings_by_parent, lookups):

  • siblings_by_parent: tables wired via dlt resolve + include_from_parent; rows carry a _<parent>_<key> column and are grouped by parent id so the caller can attach them under the sibling table's name in the children dict.
  • lookups: tables in the same dataset with no parent ref column, loaded in full and keyed by table name. Used by transforms for cross-reference data (e.g. attribution → role/post/person).

dlt derives the ref column from the extract resource the children resolve from (_<resource>_<key>). ref_parent names that resource when it differs from parent_table — e.g. when the transform reads a filtered transformer table (data_products) but its children were extracted against the broad upstream resource (data_products_all).

Sibling rows stay dlt-flattened (custom_properties__x columns): transforms address raw records by their flat keys.

Source code in src/metadataswiss_connector/dcat/dlt_schema.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def load_sibling_children(
    client,
    dataset_name: str,
    parent_table: str,
    *,
    parent_key: str,
    ref_parent: str | None = None,
) -> tuple[dict[str, dict[str, list[dict]]], dict[str, list[dict]]]:
    """Load top-level sibling tables alongside the parent.

    Returns ``(siblings_by_parent, lookups)``:

    - ``siblings_by_parent``: tables wired via dlt ``resolve`` +
      ``include_from_parent``; rows carry a ``_<parent>_<key>`` column and
      are grouped by parent id so the caller can attach them under the
      sibling table's name in the ``children`` dict.
    - ``lookups``: tables in the same dataset with no parent ref column,
      loaded in full and keyed by table name. Used by transforms for
      cross-reference data (e.g. attribution → role/post/person).

    dlt derives the ref column from the *extract* resource the children
    resolve from (``_<resource>_<key>``). ``ref_parent`` names that
    resource when it differs from ``parent_table`` — e.g. when the
    transform reads a filtered transformer table (``data_products``) but
    its children were extracted against the broad upstream resource
    (``data_products_all``).

    Sibling rows stay dlt-flattened (``custom_properties__x`` columns):
    transforms address raw records by their flat keys.
    """
    query = sql_client_query(client)
    ref_col = f"_{ref_parent or parent_table}_{parent_key}"

    result: dict[str, dict[str, list[dict]]] = {}
    lookups: dict[str, list[dict]] = {}
    for table in _tables_in_schema(query, dataset_name):
        if (
            table.startswith("_dlt")
            or table == parent_table
            or table.startswith(f"{parent_table}__")
        ):
            continue
        try:
            columns, rows = query(f'SELECT * FROM "{dataset_name}"."{table}"')
        except _UNDEFINED_TABLE_ERRORS:
            continue
        if ref_col in columns:
            rows_by_parent: dict[str, list[dict]] = defaultdict(list)
            for row in rows:
                rec = dict(zip(columns, row))
                parent_id = rec.pop(ref_col, None)
                if parent_id is None:
                    continue
                rows_by_parent[parent_id].append(_strip_dlt_columns(rec))
            if rows_by_parent:
                result[table] = dict(rows_by_parent)
        elif "__" not in table:
            recs = [_strip_dlt_columns(dict(zip(columns, row))) for row in rows]
            if recs:
                lookups[table] = recs
    return result, lookups

lookups

Cross-reference lookup tables for one transform run.

Lookups wraps the {table: [rows]} dict produced by dlt_schema.load_sibling_children (plus any synthetic tables injected by the pipeline) behind a read-only mapping with memoized per-key indexes. Transforms are called once per record but share one instance across the whole run, so by/unique_by replace repeated full-table scans (O(records × rows)) with a single index build per (table, key).

Lookups

Bases: Mapping[str, list[dict]]

Read-only view over the lookup tables of one transform run.

Source code in src/metadataswiss_connector/dcat/lookups.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class Lookups(Mapping[str, list[dict]]):
    """Read-only view over the lookup tables of one transform run."""

    def __init__(self, tables: Mapping[str, list[dict]] | None = None) -> None:
        self._tables: dict[str, list[dict]] = dict(tables or {})
        self._grouped: dict[tuple[str, str], dict[Any, list[dict]]] = {}
        self._unique: dict[tuple[str, str], dict[Any, dict]] = {}

    def __getitem__(self, table: str) -> list[dict]:
        return self._tables[table]

    def __iter__(self) -> Iterator[str]:
        return iter(self._tables)

    def __len__(self) -> int:
        return len(self._tables)

    def by(self, table: str, key: str) -> dict[Any, list[dict]]:
        """Rows of ``table`` grouped by ``row[key]`` (memoized).

        Rows without the key (or with a None value) are dropped; an
        unknown table yields an empty index.
        """
        cache_key = (table, key)
        grouped = self._grouped.get(cache_key)
        if grouped is None:
            grouped = {}
            for row in self._tables.get(table, []):
                value = row.get(key)
                if value is not None:
                    grouped.setdefault(value, []).append(row)
            self._grouped[cache_key] = grouped
        return grouped

    def unique_by(self, table: str, key: str) -> dict[Any, dict]:
        """One row of ``table`` per ``row[key]`` (memoized).

        For 1:1 reference tables (``collection_agencies``,
        ``collections``, …). Later rows win on duplicate keys, mirroring
        a dict comprehension over the raw rows.
        """
        cache_key = (table, key)
        unique = self._unique.get(cache_key)
        if unique is None:
            unique = {
                value: row
                for row in self._tables.get(table, [])
                if (value := row.get(key)) is not None
            }
            self._unique[cache_key] = unique
        return unique

by 

by(table: str, key: str) -> dict[Any, list[dict]]

Rows of table grouped by row[key] (memoized).

Rows without the key (or with a None value) are dropped; an unknown table yields an empty index.

Source code in src/metadataswiss_connector/dcat/lookups.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def by(self, table: str, key: str) -> dict[Any, list[dict]]:
    """Rows of ``table`` grouped by ``row[key]`` (memoized).

    Rows without the key (or with a None value) are dropped; an
    unknown table yields an empty index.
    """
    cache_key = (table, key)
    grouped = self._grouped.get(cache_key)
    if grouped is None:
        grouped = {}
        for row in self._tables.get(table, []):
            value = row.get(key)
            if value is not None:
                grouped.setdefault(value, []).append(row)
        self._grouped[cache_key] = grouped
    return grouped

unique_by 

unique_by(table: str, key: str) -> dict[Any, dict]

One row of table per row[key] (memoized).

For 1:1 reference tables (collection_agencies, collections, …). Later rows win on duplicate keys, mirroring a dict comprehension over the raw rows.

Source code in src/metadataswiss_connector/dcat/lookups.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def unique_by(self, table: str, key: str) -> dict[Any, dict]:
    """One row of ``table`` per ``row[key]`` (memoized).

    For 1:1 reference tables (``collection_agencies``,
    ``collections``, …). Later rows win on duplicate keys, mirroring
    a dict comprehension over the raw rows.
    """
    cache_key = (table, key)
    unique = self._unique.get(cache_key)
    if unique is None:
        unique = {
            value: row
            for row in self._tables.get(table, [])
            if (value := row.get(key)) is not None
        }
        self._unique[cache_key] = unique
    return unique

published_ids_table 

published_ids_table(resource_name: str) -> str

Name of the synthetic lookup carrying a resource's published I14Y ids.

Injected by the pipeline (see pipeline.published_id_lookups) with rows {"source_id": ..., "i14y_id": ...} so transforms can link to already-published sibling records without reading sync state.

Source code in src/metadataswiss_connector/dcat/lookups.py 
18
19
20
21
22
23
24
25
def published_ids_table(resource_name: str) -> str:
    """Name of the synthetic lookup carrying a resource's published I14Y ids.

    Injected by the pipeline (see ``pipeline.published_id_lookups``) with
    rows ``{"source_id": ..., "i14y_id": ...}`` so transforms can link to
    already-published sibling records without reading sync state.
    """
    return f"published_ids_{resource_name}"

shacl

Source-neutral SHACL/Turtle builder for I14Y dataset structure imports.

I14Y accepts a SHACL Turtle document at POST /datasets/{id}/structures/imports describing a dataset's attributes (columns). This module renders that document from a list of StructureComponent dicts — one per attribute — without knowing anything about where the components came from.

Per-source adapters (e.g. sources/dataspot/structure.py) translate their own raw rows into StructureComponent dicts and call build_shacl_turtle from here. To wire up a second source, write that adapter — no changes to this file should be needed.

The Turtle is hand-built rather than via rdflib to avoid a heavy dependency: the output shape is small, deterministic, and only needs to escape string literals correctly.

StructureComponent

Bases: TypedDict

One sh:property block worth of fields.

path_iri and xsd_datatype are required; all other keys are optional and emitted only when present. min_count/max_count follow SHACL semantics (min_count=0 means optional, max_count=1 means single-valued). conforms_to_iri is typically the I14Y concept IRI for a code-list-typed attribute.

Source code in src/metadataswiss_connector/dcat/shacl.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class StructureComponent(TypedDict, total=False):
    """One ``sh:property`` block worth of fields.

    ``path_iri`` and ``xsd_datatype`` are required; all other keys are
    optional and emitted only when present. ``min_count``/``max_count``
    follow SHACL semantics (``min_count=0`` means optional, ``max_count=1``
    means single-valued). ``conforms_to_iri`` is typically the I14Y
    concept IRI for a code-list-typed attribute.
    """

    path_iri: str
    name: str | None
    description: str | None
    xsd_datatype: str
    conforms_to_iri: str | None
    order: int | None
    min_count: int | None
    max_count: int | None
    min_inclusive: float | int | None
    max_inclusive: float | int | None

build_shacl_turtle 

build_shacl_turtle(
    *,
    subject_iri: str,
    subject_label: str | None,
    components: list[StructureComponent]
) -> str | None

Render a SHACL NodeShape for subject_iri.

Returns None when components is empty, so callers can skip the upload entirely. Component order in the input list determines block order in the output; sort upstream if you need a specific sh:order sequence.

Source code in src/metadataswiss_connector/dcat/shacl.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def build_shacl_turtle(
    *,
    subject_iri: str,
    subject_label: str | None,
    components: list[StructureComponent],
) -> str | None:
    """Render a SHACL ``NodeShape`` for ``subject_iri``.

    Returns ``None`` when ``components`` is empty, so callers can skip
    the upload entirely. Component order in the input list determines
    block order in the output; sort upstream if you need a specific
    ``sh:order`` sequence.
    """
    if not components:
        return None

    lines: list[str] = [_PREFIXES, ""]
    lines.append(f"<{subject_iri}> a sh:NodeShape ;")
    if subject_label:
        lines.append(f"    rdfs:label {_literal(subject_label)} ;")

    blocks = [_property_block(c) for c in components]
    for idx, block in enumerate(blocks):
        terminator = " ;" if idx < len(blocks) - 1 else " ."
        lines.append(f"    sh:property {block}{terminator}")

    return "\n".join(lines) + "\n"