Skip to content

API reference — core

The framework-free core: the source registry, configuration, the orchestrator-agnostic pipeline helpers and the I14Y reconciliation logic.

registry

Plugin registry for catalog sources.

Each source describes what to extract (a dlt source factory) and how to map each of its resources to an I14Y DCAT model (a transform per resource).

TransformFn

Bases: Protocol

Contract for mapping a raw record to an I14Y DCAT model.

children carries dlt-flattened 1:n child tables keyed by the child field name (e.g. "tags" → ["foo", "bar"]). Sources that don't use children can ignore the argument.

lookups is the per-run Lookups view over the source's cross-reference tables (plus pipeline-injected synthetic tables like the published-ID maps); one instance is shared across all records of a run so its memoized indexes amortise.

The return value is either the typed Pydantic model alone, or a (model, extras) tuple. extras is a JSON-serialisable dict carrying sidecar payloads that the I14Y model itself can't hold — e.g. code-list entries, which I14Y publishes via a separate bulk endpoint after the parent concept is created. Extras are stored alongside the transformed record and replayed by sync as a post-create/update follow-up call.

Source code in src/metadataswiss_connector/registry.py 
24
25
26
27
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
class TransformFn(Protocol):
    """Contract for mapping a raw record to an I14Y DCAT model.

    ``children`` carries dlt-flattened 1:n child tables keyed by the
    child field name (e.g. ``"tags"`` → ``["foo", "bar"]``). Sources
    that don't use children can ignore the argument.

    ``lookups`` is the per-run ``Lookups`` view over the source's
    cross-reference tables (plus pipeline-injected synthetic tables like
    the published-ID maps); one instance is shared across all records of
    a run so its memoized indexes amortise.

    The return value is either the typed Pydantic model alone, or a
    ``(model, extras)`` tuple. ``extras`` is a JSON-serialisable dict
    carrying sidecar payloads that the I14Y model itself can't hold —
    e.g. code-list entries, which I14Y publishes via a separate bulk
    endpoint after the parent concept is created. Extras are stored
    alongside the transformed record and replayed by sync as a
    post-create/update follow-up call.
    """

    def __call__(
        self,
        record: dict,
        children: dict[str, list],
        *,
        lookups: Lookups,
        publisher: str,
    ) -> BaseModel | tuple[BaseModel, dict]: ...

ResourceSpec dataclass

How one dlt resource is mapped + published to I14Y.

Attributes:

Name Type Description
transform TransformFn

Function mapping a raw record to an I14Y input model.
kind ResourceKind

Which I14Y entity the model targets — controls sync routing.
lookups tuple[str, ...]

Names of other raw dlt resources (in the same source) that this resource's transform reads via the children or lookups argument. Declared explicitly so the Dagster lineage shows these cross-table reads as upstream deps on the transformed asset; runtime behaviour is unaffected (lookups are still discovered dynamically by load_sibling_children).
sibling_parent str | None

Name of the extract dlt resource that this resource's sibling child tables resolve from. dlt names their parent-ref column _<extract resource>_id, so this must be set whenever the transform reads a filtered transformer table whose children were extracted against a differently-named upstream resource (e.g. data_products reading distributions extracted via data_products_all). None means the children reference this resource's own name.
Source code in src/metadataswiss_connector/registry.py 
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
@dataclass(frozen=True)
class ResourceSpec:
    """How one dlt resource is mapped + published to I14Y.

    Attributes:
        transform: Function mapping a raw record to an I14Y input model.
        kind: Which I14Y entity the model targets — controls sync routing.
        lookups: Names of *other* raw dlt resources (in the same source)
                that this resource's transform reads via the ``children``
                or ``lookups`` argument. Declared explicitly so the
                Dagster lineage shows these cross-table reads as
                upstream deps on the transformed asset; runtime behaviour
                is unaffected (lookups are still discovered dynamically
                by ``load_sibling_children``).
        sibling_parent: Name of the *extract* dlt resource that this
                resource's sibling child tables resolve from. dlt names
                their parent-ref column ``_<extract resource>_id``, so
                this must be set whenever the transform reads a filtered
                transformer table whose children were extracted against
                a differently-named upstream resource (e.g.
                ``data_products`` reading ``distributions`` extracted
                via ``data_products_all``). ``None`` means the children
                reference this resource's own name.
    """

    transform: TransformFn
    kind: ResourceKind = "dataset"
    lookups: tuple[str, ...] = ()
    sibling_parent: str | None = None

CatalogSource dataclass

A registered catalog source.

Attributes:

Name Type Description
name str

Stable identifier; also used as dlt pipeline name.
dlt_source_factory Callable[[], DltSource]

Zero-arg callable returning a configured DltSource. A factory (not an instance) lets each run build a fresh source with current credentials.
resources dict[str, ResourceSpec]

Maps dlt resource name → ResourceSpec (transform + kind).
publisher str | None

Optional publisher override. If None, the caller's global publisher (e.g. from I14YConfig) is used.
drain_extract_skips Callable[[], list[dict]] | None

Optional zero-arg callable returning (and clearing) the items the source skipped during the last extract (e.g. endpoints answering with server-side errors). Each dict must match the invalid_details shape consumed by the quality-issues email, plus a resource key naming the dlt resource it belongs to.
Source code in src/metadataswiss_connector/registry.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@dataclass(frozen=True)
class CatalogSource:
    """A registered catalog source.

    Attributes:
        name: Stable identifier; also used as dlt pipeline name.
        dlt_source_factory: Zero-arg callable returning a configured
              ``DltSource``. A factory (not an instance) lets each run
              build a fresh source with current credentials.
        resources: Maps dlt resource name → ResourceSpec (transform + kind).
        publisher: Optional publisher override. If ``None``, the caller's
              global publisher (e.g. from ``I14YConfig``) is used.
        drain_extract_skips: Optional zero-arg callable returning (and
              clearing) the items the source skipped during the last
              extract (e.g. endpoints answering with server-side errors).
              Each dict must match the ``invalid_details`` shape consumed
              by the quality-issues email, plus a ``resource`` key naming
              the dlt resource it belongs to.
    """

    name: str
    dlt_source_factory: Callable[[], DltSource]
    resources: dict[str, ResourceSpec]
    publisher: str | None = None
    drain_extract_skips: Callable[[], list[dict]] | None = None

run_source 

run_source(
    source: CatalogSource,
    *,
    publisher: str,
    destination=None
) -> None

Extract + transform a single source end-to-end (no publish).

Source code in src/metadataswiss_connector/registry.py 
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
def run_source(
    source: CatalogSource,
    *,
    publisher: str,
    destination=None,
) -> None:
    """Extract + transform a single source end-to-end (no publish)."""
    # Deferred imports: registry sits below the orchestration layer
    # (pipeline imports registry), so pull these in lazily.
    from metadataswiss_connector.pipeline import published_id_lookups
    from metadataswiss_connector.resources import duckdb_destination

    destination = destination or duckdb_destination()
    effective_publisher = source.publisher or publisher

    raw_pipeline = dlt.pipeline(
        pipeline_name=source.name,
        destination=destination,
        dataset_name=f"{source.name}_raw",
    )
    raw_pipeline.run(source.dlt_source_factory())

    run_transform(
        pipeline_raw=raw_pipeline,
        resources=source.resources,
        destination=destination,
        publisher=effective_publisher,
        extra_lookups=published_id_lookups(source),
    )

config

Framework-free configuration objects and env accessors.

These live outside of the orchestrator so the connector can be used as a plain library (CLI, tests, notebooks). All environment access of the connector is gathered here (the dlt source credentials are the exception — dlt injects those itself); the accessors read at call time, so containerised deployments and tests can override via the environment without import-order concerns.

I14YConfig

Bases: BaseModel

Connection + publisher config for the I14Y Partner API.

Source code in src/metadataswiss_connector/config.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class I14YConfig(BaseModel):
    """Connection + publisher config for the I14Y Partner API."""

    base_url: str
    token_url: str
    client_id: str
    client_secret: str
    publisher: str
    user_agent: str

    @classmethod
    def from_env(cls) -> "I14YConfig":
        return cls(
            base_url=os.environ["I14Y_BASE_URL"],
            token_url=os.environ["I14Y_TOKEN_URL"],
            client_id=os.environ["I14Y_CLIENT_ID"],
            client_secret=os.environ["I14Y_CLIENT_SECRET"],
            publisher=os.environ["I14Y_PUBLISHER_IDENTIFIER"],
            user_agent=os.environ["I14Y_USER_AGENT"],
        )

state_dir 

state_dir() -> Path

Directory holding the per-resource remote-ID state files.

Override via CONNECTOR_DATA_DIR to put them on a persistent volume in a containerised deployment; defaults to data/state/ in the working directory for local use.

Source code in src/metadataswiss_connector/config.py 
17
18
19
20
21
22
23
24
def state_dir() -> Path:
    """Directory holding the per-resource remote-ID state files.

    Override via ``CONNECTOR_DATA_DIR`` to put them on a persistent
    volume in a containerised deployment; defaults to ``data/state/``
    in the working directory for local use.
    """
    return Path(os.environ.get("CONNECTOR_DATA_DIR", "data/state"))

duckdb_path 

duckdb_path() -> str

Location of the local DuckDB warehouse.

Override via DUCKDB_PATH to point at a persistent volume in a containerised deployment; defaults to a file in the working directory for local/CLI use.

Source code in src/metadataswiss_connector/config.py 
27
28
29
30
31
32
33
34
def duckdb_path() -> str:
    """Location of the local DuckDB warehouse.

    Override via ``DUCKDB_PATH`` to point at a persistent volume in a
    containerised deployment; defaults to a file in the working
    directory for local/CLI use.
    """
    return os.environ.get("DUCKDB_PATH", "data/metadata.duckdb")

dataspot_web_base 

dataspot_web_base() -> tuple[str, str] | None

(base_url, database_name) of the dataspot UI, or None if unset.

Reuses the dlt source-connection env vars SOURCES__DATASPOT__* so deep links (e.g. in alert emails) point at the same instance the data was extracted from.

Source code in src/metadataswiss_connector/config.py 
37
38
39
40
41
42
43
44
45
46
47
48
def dataspot_web_base() -> tuple[str, str] | None:
    """(base_url, database_name) of the dataspot UI, or ``None`` if unset.

    Reuses the dlt source-connection env vars ``SOURCES__DATASPOT__*``
    so deep links (e.g. in alert emails) point at the same instance the
    data was extracted from.
    """
    base = os.environ.get("SOURCES__DATASPOT__BASE_URL")
    database = os.environ.get("SOURCES__DATASPOT__DATABASE_NAME")
    if not base or not database:
        return None
    return base.rstrip("/"), database

pipeline

Framework-free per-source / per-resource pipeline helpers.

These wrap the three logical steps (extract, transform, publish) so they can be invoked from the CLI and from an orchestrator (e.g. Dagster) alike. The CLI runs them in source-wide loops; Dagster materialises one resource at a time.

state_path 

state_path(source_name: str, resource_name: str) -> Path

One state file per (source, resource) pair (under config.state_dir()).

Source code in src/metadataswiss_connector/pipeline.py 
29
30
31
32
33
def state_path(source_name: str, resource_name: str) -> Path:
    """One state file per (source, resource) pair (under ``config.state_dir()``)."""
    directory = state_dir()
    directory.mkdir(parents=True, exist_ok=True)
    return directory / f"{source_name}_{resource_name}_ids.json"

extract_source 

extract_source(source: CatalogSource) -> None

Run the dlt extract for an entire source (all resources at once).

Source code in src/metadataswiss_connector/pipeline.py 
36
37
38
39
40
41
42
43
44
def extract_source(source: CatalogSource) -> None:
    """Run the dlt extract for an entire source (all resources at once)."""
    pipeline = raw_pipeline_for(source)
    load_info = pipeline.run(source.dlt_source_factory())
    row_counts = pipeline.last_trace.last_normalize_info.row_counts
    for table, count in sorted(row_counts.items()):
        if not table.startswith("_dlt"):
            logger.info("extracted %s rows into %s", count, table)
    logger.info("%s", load_info)

published_id_lookups 

published_id_lookups(
    source: CatalogSource,
) -> dict[str, list[dict]]

Synthetic lookup tables exposing each resource's published I14Y ids.

One table per resource of source (omitted while nothing is published yet), named via published_ids_table and carrying rows {"source_id": ..., "i14y_id": ...} from the resource's sync state. Injected into transforms so they can reference already- published sibling records (e.g. a dataservice's servesDatasets) without reading sync state themselves.

Resolution is against the last persisted state: a brand-new dataset+API pair links on the run after the dataset is first published. There is no within-run ordering guarantee between the independent per-resource publish steps anyway, so this is the consistent choice.

Source code in src/metadataswiss_connector/pipeline.py 
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
def published_id_lookups(source: CatalogSource) -> dict[str, list[dict]]:
    """Synthetic lookup tables exposing each resource's published I14Y ids.

    One table per resource of ``source`` (omitted while nothing is
    published yet), named via ``published_ids_table`` and carrying rows
    ``{"source_id": ..., "i14y_id": ...}`` from the resource's sync
    state. Injected into transforms so they can reference already-
    published sibling records (e.g. a dataservice's ``servesDatasets``)
    without reading sync state themselves.

    Resolution is against the last *persisted* state: a brand-new
    dataset+API pair links on the run after the dataset is first
    published. There is no within-run ordering guarantee between the
    independent per-resource publish steps anyway, so this is the
    consistent choice.
    """
    lookups: dict[str, list[dict]] = {}
    for resource_name in source.resources:
        id_map = active_id_map(state_path(source.name, resource_name))
        if id_map:
            lookups[published_ids_table(resource_name)] = [
                {"source_id": source_id, "i14y_id": i14y_id}
                for source_id, i14y_id in sorted(id_map.items())
            ]
    return lookups

transform_resource 

transform_resource(
    source: CatalogSource, resource_name: str
) -> dict

Transform raw rows of a single resource into the i14y_dcat dataset.

Returns the stats dict from run_transform_resource so Dagster assets (and any other caller) can surface row counts and validation rate as materialization metadata.

Source code in src/metadataswiss_connector/pipeline.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def transform_resource(source: CatalogSource, resource_name: str) -> dict:
    """Transform raw rows of a single resource into the i14y_dcat dataset.

    Returns the stats dict from ``run_transform_resource`` so Dagster
    assets (and any other caller) can surface row counts and validation
    rate as materialization metadata.
    """
    spec = source.resources[resource_name]
    cfg = I14YConfig.from_env()
    publisher = source.publisher or cfg.publisher
    return run_transform_resource(
        pipeline_raw=raw_pipeline_for(source),
        resource_name=resource_name,
        spec=spec,
        destination=duckdb_destination(),
        publisher=publisher,
        extra_lookups=published_id_lookups(source),
    )

publish_resource 

publish_resource(
    source: CatalogSource,
    resource_name: str,
    *,
    limit: int | None = None,
    log: Logger | None = None
) -> SyncResult

Publish transformed records of a single resource to I14Y.

log is threaded into the I14Y client so per-request telemetry (method, status, duration, request id) shows up on the Dagster asset's run log when called from an asset.

Source code in src/metadataswiss_connector/pipeline.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def publish_resource(
    source: CatalogSource,
    resource_name: str,
    *,
    limit: int | None = None,
    log: logging.Logger | None = None,
) -> SyncResult:
    """Publish transformed records of a single resource to I14Y.

    ``log`` is threaded into the I14Y client so per-request telemetry
    (method, status, duration, request id) shows up on the Dagster
    asset's run log when called from an asset.
    """
    spec = source.resources[resource_name]
    records = read_transformed(resource_name, limit=limit)
    if not records:
        (log or logger).warning(
            "no transformed records for %s.%s — skipping",
            source.name, resource_name,
        )
        return SyncResult()
    with i14y_client_from_env(logger=log) as client:
        return sync(
            client,
            records,
            kind=spec.kind,
            state_path=state_path(source.name, resource_name),
        )

sync

Sync logic: reconcile local I14Y input records with remote state.

Generic core (sync_records / purge_records) reconciles a list of transformed records against a per-resource state file via any I14Y resource client that exposes the standard CRUD + lifecycle operations. The public sync / purge entrypoints dispatch on a kind string ("dataset" / "concept") via the _KINDS registry.

Change detection is hash-based: each transformed record is canonically serialised and hashed; the digest is persisted alongside the I14Y id and compared on the next run. Any change to the transform logic, the source data, or upstream lookups that affects the published payload flips the hash and forces a re-publish — no manual version bumps required.

ResourceClient

Bases: Protocol

Duck-typed contract satisfied by client.datasets / client.concepts.

Source code in src/metadataswiss_connector/sync.py 
40
41
42
43
44
45
46
47
class ResourceClient(Protocol):
    """Duck-typed contract satisfied by ``client.datasets`` / ``client.concepts``."""

    def create(self, model: BaseModel) -> UUID: ...
    def update(self, id_: UUID | str, model: BaseModel) -> None: ...
    def publish_initial(self, id_: UUID | str) -> None: ...
    def decommission_and_delete(self, id_: UUID | str) -> None: ...
    def apply_extras(self, id_: UUID | str, extras: dict) -> None: ...

FailedRecord dataclass

A record that could not be published to I14Y.

Carries the source identifier (dataspot UUID), the error text, and — when available — a human-readable title so the alert email can name the record instead of only showing its opaque UUID.

Source code in src/metadataswiss_connector/sync.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class FailedRecord:
    """A record that could not be published to I14Y.

    Carries the source identifier (dataspot UUID), the error text, and —
    when available — a human-readable title so the alert email can name
    the record instead of only showing its opaque UUID.
    """

    source_id: str
    error: str
    title: str | None = None

    def __str__(self) -> str:
        label = f"{self.title} " if self.title else ""
        return f"{label}[{self.source_id}]: {self.error}"

SyncResult dataclass

Summary of a sync run.

Source code in src/metadataswiss_connector/sync.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@dataclass
class SyncResult:
    """Summary of a sync run."""

    created: list[str] = field(default_factory=list)
    updated: list[str] = field(default_factory=list)
    unchanged: list[str] = field(default_factory=list)
    deleted: list[str] = field(default_factory=list)
    failed: list[FailedRecord] = field(default_factory=list)

    def summary(self) -> str:
        parts = []
        if self.created:
            parts.append(f"{len(self.created)} created")
        if self.updated:
            parts.append(f"{len(self.updated)} updated")
        if self.unchanged:
            parts.append(f"{len(self.unchanged)} unchanged")
        if self.deleted:
            parts.append(f"{len(self.deleted)} deleted")
        if self.failed:
            parts.append(f"{len(self.failed)} failed")
        return ", ".join(parts) or "no changes"

SyncState

Persistent mapping of source identifier → I14Y UUID + payload hash.

Stored as a JSON file so the sync knows which records it owns on I14Y without having to query the remote API, and can detect changes to the transformed payload without re-publishing untouched records.

Source code in src/metadataswiss_connector/sync.py 
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
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
class SyncState:
    """Persistent mapping of source identifier → I14Y UUID + payload hash.

    Stored as a JSON file so the sync knows which records it owns on
    I14Y without having to query the remote API, and can detect changes
    to the transformed payload without re-publishing untouched records.
    """

    def __init__(self, path: Path) -> None:
        self.path = path
        self._data: dict[str, dict] = {}
        self._load()

    def _load(self) -> None:
        if self.path.exists():
            with open(self.path) as f:
                self._data = json.load(f)
            logger.info("Loaded %d entries from state file", len(self._data))
        else:
            logger.info("No state file found at %s, starting fresh", self.path)

    def save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.path, "w") as f:
            json.dump(self._data, f, indent=2)
        logger.info("Saved %d entries to state file", len(self._data))

    def get_i14y_id(self, source_id: str) -> str | None:
        entry = self._data.get(source_id)
        if not entry or entry.get("deleted_at"):
            return None
        return entry.get("id")

    def get_payload_hash(self, source_id: str) -> str | None:
        entry = self._data.get(source_id)
        if not entry or entry.get("deleted_at"):
            return None
        return entry.get("payload_hash")

    def contains(self, source_id: str) -> bool:
        entry = self._data.get(source_id)
        return entry is not None and not entry.get("deleted_at")

    def add(
        self,
        source_id: str,
        i14y_id: str,
        payload_hash: str,
    ) -> None:
        existing = self._data.get(source_id) or {}
        # Re-creation after a soft-delete: the I14Y UUID is new, so
        # created_at must reflect the current incarnation, not the
        # previous one.
        is_recreation = bool(existing.get("deleted_at"))
        now = datetime.now(timezone.utc).isoformat()
        if is_recreation or not existing.get("created_at"):
            created_at = now
        else:
            created_at = existing["created_at"]
        self._data[source_id] = {
            "id": i14y_id,
            "payload_hash": payload_hash,
            "created_at": created_at,
            "updated_at": now,
        }

    def remove(self, source_id: str) -> None:
        entry = self._data.get(source_id)
        if not entry:
            return
        entry["deleted_at"] = datetime.now(timezone.utc).isoformat()

    def all_source_ids(self) -> set[str]:
        return {sid for sid, entry in self._data.items() if not entry.get("deleted_at")}

payload_hash 

payload_hash(record: dict) -> str

SHA-256 over a canonical JSON serialisation of the full record.

default=str coerces datetimes (and any other non-JSON-native values dlt may surface) deterministically. sort_keys ensures dict ordering doesn't perturb the digest. Extras are included because they're part of the published payload.

Source code in src/metadataswiss_connector/sync.py 
58
59
60
61
62
63
64
65
66
67
def payload_hash(record: dict) -> str:
    """SHA-256 over a canonical JSON serialisation of the full record.

    ``default=str`` coerces datetimes (and any other non-JSON-native
    values dlt may surface) deterministically. ``sort_keys`` ensures
    dict ordering doesn't perturb the digest. Extras are included
    because they're part of the published payload.
    """
    canonical = json.dumps(record, sort_keys=True, default=str, ensure_ascii=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

active_id_map 

active_id_map(state_path: Path) -> dict[str, str]

Source identifier → I14Y id for every record currently active on I14Y.

Reads the given per-resource state file; soft-deleted entries and entries without an I14Y id (never successfully created) are excluded. Empty when the state file doesn't exist yet (nothing published). Owned by this module so no other layer needs to know the state-file format.

Source code in src/metadataswiss_connector/sync.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def active_id_map(state_path: Path) -> dict[str, str]:
    """Source identifier → I14Y id for every record currently active on I14Y.

    Reads the given per-resource state file; soft-deleted entries and
    entries without an I14Y id (never successfully created) are excluded.
    Empty when the state file doesn't exist yet (nothing published).
    Owned by this module so no other layer needs to know the state-file
    format.
    """
    if not state_path.exists():
        return {}
    state = SyncState(state_path)
    return {
        source_id: i14y_id
        for source_id in sorted(state.all_source_ids())
        if (i14y_id := state.get_i14y_id(source_id))
    }

sync_records 

sync_records(
    resource: ResourceClient,
    source_records: list[dict],
    *,
    model_class: type[BaseModel],
    source_id_for: SourceIdFn,
    state_path: Path,
    dry_run: bool = False,
    force: bool = False
) -> SyncResult

Synchronise transformed records to I14Y via the given resource client.

Parameters:

Name Type Description Default
resource ResourceClient

I14Y resource accessor (client.datasets, client.concepts, …) implementing ResourceClient.

 required
source_records list[dict]

dicts produced by the transform step, each validating as model_class.

 required
model_class type[BaseModel]

Pydantic input model class for the resource.

 required
source_id_for SourceIdFn

Pull the stable source identifier out of a record. All current kinds use identifiers[0].

 required
state_path Path

Per-resource state file (one file per source+resource).

 required
dry_run bool

Compute the diff without calling the API.

 False
force bool

Re-publish existing records regardless of payload hash.

 False
Source code in src/metadataswiss_connector/sync.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def sync_records(
    resource: ResourceClient,
    source_records: list[dict],
    *,
    model_class: type[BaseModel],
    source_id_for: SourceIdFn,
    state_path: Path,
    dry_run: bool = False,
    force: bool = False,
) -> SyncResult:
    """Synchronise transformed records to I14Y via the given resource client.

    Args:
        resource: I14Y resource accessor (``client.datasets``,
                 ``client.concepts``, …) implementing ``ResourceClient``.
        source_records: dicts produced by the transform step, each
                       validating as ``model_class``.
        model_class: Pydantic input model class for the resource.
        source_id_for: Pull the stable source identifier out of a record.
                       All current kinds use ``identifiers[0]``.
        state_path: Per-resource state file (one file per source+resource).
        dry_run: Compute the diff without calling the API.
        force: Re-publish existing records regardless of payload hash.
    """
    state = SyncState(state_path)
    source_by_id = _build_source_index(source_records, source_id_for)
    hash_by_id = {sid: payload_hash(rec) for sid, rec in source_by_id.items()}
    to_create, to_check, to_delete = _compute_diff(set(source_by_id), state.all_source_ids())

    logger.info(
        "Sync plan: %d new, %d to check for updates, %d to delete",
        len(to_create), len(to_check), len(to_delete),
    )

    if dry_run:
        return _dry_run_result(
            to_create, to_check, to_delete, hash_by_id, state, force=force,
        )

    run = _SyncRun(resource=resource, state=state, model_class=model_class)
    # try/finally guarantees state is persisted even on unexpected errors,
    # so successfully-created I14Y records aren't orphaned by a crash.
    try:
        for source_id in sorted(to_create):
            run.create(source_id, source_by_id[source_id], hash_by_id[source_id])
        for source_id in sorted(to_check):
            run.update_if_changed(
                source_id, source_by_id[source_id], hash_by_id[source_id],
                force=force,
            )
        for source_id in sorted(to_delete):
            run.delete(source_id)
    finally:
        state.save()

    return run.result

sync 

sync(
    client: I14YClient,
    source_records: list[dict],
    *,
    kind: str,
    state_path: Path,
    dry_run: bool = False,
    force: bool = False
) -> SyncResult

Sync transformed records to I14Y for the given resource kind.

Source code in src/metadataswiss_connector/sync.py 
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
def sync(
    client: I14YClient,
    source_records: list[dict],
    *,
    kind: str,
    state_path: Path,
    dry_run: bool = False,
    force: bool = False,
) -> SyncResult:
    """Sync transformed records to I14Y for the given resource ``kind``."""
    resource, spec = _resource_for(client, kind)
    return sync_records(
        resource,
        source_records,
        model_class=spec.model_class,
        source_id_for=spec.source_id_for,
        state_path=state_path,
        dry_run=dry_run,
        force=force,
    )

purge_records 

purge_records(
    resource: ResourceClient,
    *,
    state_path: Path,
    dry_run: bool = False
) -> SyncResult

Delete every active record recorded in the state file from I14Y.

Active = not yet marked deleted_at. Soft-deleted entries are skipped (already considered gone on I14Y per our records). On success, entries are soft-deleted in the state file.

Source code in src/metadataswiss_connector/sync.py 
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
def purge_records(
    resource: ResourceClient,
    *,
    state_path: Path,
    dry_run: bool = False,
) -> SyncResult:
    """Delete every active record recorded in the state file from I14Y.

    Active = not yet marked ``deleted_at``. Soft-deleted entries are
    skipped (already considered gone on I14Y per our records).
    On success, entries are soft-deleted in the state file.
    """
    state = SyncState(state_path)
    source_ids = sorted(state.all_source_ids())

    logger.warning("PURGE: %d active records will be deleted from I14Y", len(source_ids))

    if dry_run:
        result = SyncResult()
        result.deleted = source_ids
        return result

    run = _SyncRun(resource=resource, state=state)
    try:
        for source_id in source_ids:
            run.delete(source_id)
    finally:
        state.save()
    return run.result

purge 

purge(
    client: I14YClient,
    *,
    kind: str,
    state_path: Path,
    dry_run: bool = False
) -> SyncResult

Delete every active record of the given resource kind from I14Y.

Source code in src/metadataswiss_connector/sync.py 
546
547
548
549
550
551
552
553
554
555
def purge(
    client: I14YClient,
    *,
    kind: str,
    state_path: Path,
    dry_run: bool = False,
) -> SyncResult:
    """Delete every active record of the given resource ``kind`` from I14Y."""
    resource, _ = _resource_for(client, kind)
    return purge_records(resource, state_path=state_path, dry_run=dry_run)

resources

Shared dlt + I14Y client helpers.

duckdb_destination 

duckdb_destination()

Shared DuckDB destination used by raw and DCAT pipelines.

Source code in src/metadataswiss_connector/resources.py 
15
16
17
def duckdb_destination():
    """Shared DuckDB destination used by raw and DCAT pipelines."""
    return dlt.destinations.duckdb(duckdb_path())

raw_pipeline_for 

raw_pipeline_for(source: CatalogSource) -> dlt.Pipeline

The dlt pipeline that lands raw data for a given source in DuckDB.

Source code in src/metadataswiss_connector/resources.py 
20
21
22
23
24
25
26
def raw_pipeline_for(source: "CatalogSource") -> dlt.Pipeline:
    """The dlt pipeline that lands raw data for a given source in DuckDB."""
    return dlt.pipeline(
        pipeline_name=source.name,
        destination=duckdb_destination(),
        dataset_name=f"{source.name}_raw",
    )

i14y_client_from_env 

i14y_client_from_env(
    logger: Logger | None = None,
) -> I14YClient

Build an I14Y API client from environment-derived config.

logger lets the caller (e.g. a Dagster asset passing context.log) capture per-request telemetry alongside their own output. Defaults to the package logger.

Source code in src/metadataswiss_connector/resources.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def i14y_client_from_env(logger: logging.Logger | None = None) -> I14YClient:
    """Build an I14Y API client from environment-derived config.

    ``logger`` lets the caller (e.g. a Dagster asset passing
    ``context.log``) capture per-request telemetry alongside their own
    output. Defaults to the package logger.
    """
    cfg = I14YConfig.from_env()
    auth = I14YAuth(
        token_url=cfg.token_url,
        client_id=cfg.client_id,
        client_secret=cfg.client_secret,
    )
    kwargs: dict = {
        "base_url": cfg.base_url,
        "auth": auth,
        "user_agent": cfg.user_agent,
    }
    if logger is not None:
        kwargs["logger"] = logger
    return I14YClient(**kwargs)