chg(coat): coat pipeline

Co-authored-by: Francesco Frassinelli <francesco.frassinelli@nina.no>

Author: taheeraahmed

src/datasync/coat.py

@@ -0,0 +1,194 @@
+import json
+from datetime import datetime
+from urllib.parse import urlparse
+
+import dlt
+import typer
+from dlt.sources.helpers.rest_client import RESTClient
+from dlt.sources.helpers.rest_client.paginators import OffsetPaginator
+
+from .settings import log
+
+app = typer.Typer(help="Export COAT CKAN packages to Parquet")
+
+PAGE_SIZE = 100
+BASE_URL = "https://data.coat.no"
+ENDPOINT_URL = BASE_URL + "/api/3/action"
+
+MIXED = ("author_email", "maintainer", "maintainer_email")
+SPLIT = (
+    "associated_parties",
+    "datasets",
+    "funding",
+    "location",
+    "persons",
+    "protocol",
+    "scientific_name",
+)
+DATE_FIELDS = ("embargo", "temporal_start", "temporal_end")
+
+
+def normalize_record(record, domain: str = BASE_URL):
+    """Normalize a package record in-place."""
+    org = record.get("organization") or {}
+
+    # Handle mixed fields (can be string or list)
+    for field in MIXED:
+        value = record.get(field)
+        if isinstance(value, list):
+            record[field] = json.dumps(value)
+        elif isinstance(value, str) and value.startswith("["):
+            try:
+                parsed = json.loads(value)
+                if isinstance(parsed, list) and len(parsed) == 1:
+                    record[field] = parsed[0]
+            except json.JSONDecodeError:
+                log.error(f"Failed to parse JSON for field {field}: {value}")
+                raise ValueError(
+                    f"Failed to parse JSON for field {field}: {value}"
+                ) from None
+
+    # Split comma-separated fields into arrays
+    for field in SPLIT:
+        value = record.get(field)
+        record[field] = [v.strip() for v in value.split(",")] if value else []
+
+    # Parse date fields
+    for field in DATE_FIELDS:
+        value = record.get(field)
+        if value:
+            try:
+                record[field] = datetime.strptime(value, "%Y-%m-%d").date()
+            except (ValueError, TypeError):
+                pass
+
+    # Extract organization info
+    record["organization_name"] = org.get("name")
+    record["organization_title"] = org.get("title")
+
+    # Flatten extras to JSON object
+    record["extras"] = {e["key"]: e["value"] for e in record.get("extras", [])}
+
+    # Extract tag names
+    record["tags"] = [t["name"] for t in record.get("tags", [])]
+
+    # Build URL
+    name = record.get("name")
+    if not name:
+        log.warning(
+            f"Record {record.get('id')!r} is missing 'name', URL will be incomplete"
+        )
+    record["url"] = f"{domain}/dataset/{name or ''}"
+
+    return record
+
+
+@dlt.resource(
+    name="packages",
+    primary_key="id",
+    write_disposition="replace",
+)
+def packages(
+    api_key: str = "",
+    base_url: str = ENDPOINT_URL,
+):
+    """Fetch packages from CKAN API and normalize them."""
+    log.info(f"Starting package extraction from {base_url}")
+
+    parsed = urlparse(base_url)
+    if not parsed.scheme or not parsed.netloc:
+        log.error("Invalid base_url: %s", base_url)
+        raise ValueError(f"Invalid base_url: {base_url!r}")
+    domain = f"{parsed.scheme}://{parsed.netloc}"
+
+    client = RESTClient(
+        base_url=base_url,
+        headers={"Authorization": api_key, "Accept": "application/json"},
+    )
+
+    paginator = OffsetPaginator(
+        limit=PAGE_SIZE,
+        offset_param="start",
+        limit_param="rows",
+        total_path="result.count",
+    )
+
+    total_records = 0
+    for page in client.paginate(
+        "/ckan_package_search",
+        params={"include_private": "true"},
+        paginator=paginator,
+    ):
+        for record in page:
+            yield normalize_record(record, domain=domain)
+            total_records += 1
+
+    log.info(f"Extraction complete. Total records: {total_records}")
+
+
+@dlt.transformer(
+    data_from=packages,
+    name="resources",
+    primary_key="id",
+    write_disposition="replace",
+)
+def resources(pkg):
+    """Extract resources from packages."""
+    for res in pkg.get("resources", []):
+        yield {
+            **res,
+            "package_name": pkg.get("name"),
+            "package_id": pkg.get("id"),
+        }
+
+
+@dlt.source(name="coat", max_table_nesting=0)
+def coat_source(api_key: str = "", base_url: str = ENDPOINT_URL):
+    """Define the COAT data source."""
+    return packages(api_key=api_key, base_url=base_url), resources
+
+
+@app.command()
+def get_packages_and_resources(
+    bucket_url: str = typer.Option(
+        default="data",
+        envvar="COAT_BUCKET_URL",
+        help="Destination bucket URL (local path or s3://...)",
+    ),
+    api_key: str = typer.Option(
+        ...,
+        envvar="COAT_API_KEY",
+        help="CKAN API key",
+    ),
+    base_url: str = typer.Option(
+        default="https://data.coat.no/api/3/action",
+        envvar="COAT_BASE_URL",
+        help="COAT CKAN API base URL",
+    ),
+    dataset_name: str = typer.Option(
+        default="coat",
+        envvar="COAT_DATASET_NAME",
+        help="Dataset name for the extracted data",
+    ),
+):
+    """Run the COAT extraction pipeline."""
+    pipeline = dlt.pipeline(
+        pipeline_name=dataset_name,
+        destination=dlt.destinations.filesystem(
+            bucket_url=bucket_url, layout="{table_name}.{ext}"
+        ),
+        dataset_name=dataset_name,
+    )
+    run = pipeline.run(
+        coat_source(api_key=api_key, base_url=base_url), loader_file_format="parquet"
+    )
+    log.debug(
+        f"COAT pipeline output written to:\n"
+        f"- {bucket_url}/coat/resources.parquet\n"
+        f"- {bucket_url}/coat/packages.parquet"
+    )
+    log.info(f"Pipeline run completed. Load info: {run}")
+
+
+if __name__ == "__main__":
+    app()

src/datasync/main.py

@@ -5,6 +5,7 @@
 import typer
 
 from . import (
+    coat,
     dms,
     gbif_backbone,
     grass,
@@ -19,6 +20,7 @@
 app = typer.Typer(
     help="Provide subcommands for synchronizing different resources, see subcommands"
 )
+app.add_typer(coat.app, name="coat")
 app.add_typer(nva.app, name="nva")
 app.add_typer(ubw.app, name="ubw")
 app.add_typer(dms.app, name="dms")