diff --git a/API.md b/API.md index 4146278..093ec54 100644 --- a/API.md +++ b/API.md @@ -78,6 +78,8 @@ token (except under `anonymous`). | GET | `/datastore/api/v2/dump/query` | Download the result of a SQL `SELECT` as a file | | GET | `/datastore/api/v2/datastore_info` | Schema + row stats for a resource | | GET | `/datastore/api/v2/dump/{resource_id}` | Download a whole resource (CSV/NDJSON/Parquet) | +| GET | `/datastore/api/v2/dump/{resource_id}/dictionary` | Download a resource's data dictionary (CSV/JSON) | +| GET | `/datastore/api/v2/dump/{resource_id}/metadata` | Download a resource's + its dataset's metadata (CSV/JSON) | | GET | `/datastore/api/health` · `/datastore/api/ready` | Liveness / readiness | --- @@ -422,6 +424,54 @@ by this URL. --- +## `GET /datastore/api/v2/dump/{resource_id}/dictionary` + +Download the resource's data dictionary. `?format=csv` (default) or `json`. +Unlike the table dump this is served directly (`200`, no redirect) — the file +is small and built in memory. + +- **json** — the Frictionless Table Schema, exactly as `datastore_info` + returns it under `result.schema` (`application/json`). +- **csv** — one row per column, one column per field property: `name`, + `type`, `title`, `description` first, then any other property in the order + first seen. Nested values (e.g. `constraints`) are written as compact JSON + in their cell (`text/csv; charset=utf-8`). + +Named `_data_dictionary.`, where `` follows the table +dump's naming (uploaded filename, else `download_filename`, else the resource +id). Requires `read` permission on the resource. + +--- + +## `GET /datastore/api/v2/dump/{resource_id}/metadata` + +Download the resource's metadata together with its dataset's. `?format=csv` +(default) or `json`. Served directly (`200`). **Requires `AUTH_TYPE=ckan`** +(`Validation Error` otherwise) and `read` permission on the resource. + +| Field | Source | +|---|---| +| Dataset name / id / description | package `title` / `id` / `notes` | +| Data group | package `organization.title` | +| Dataset update frequency | package `update_frequency` | +| Dataset created / last updated | package `metadata_created` / `metadata_modified` | +| Licence | package `license_title` (else `license_id`) | +| Tags | package tag names | +| Resource name / id / format | resource `name` / `id` / `format` | +| Resource created | resource `created` | +| Resource last updated | resource `metadata_modified`, else `last_modified` (what the portal shows) | +| Download URL | this service's `dump/{resource_id}` URL | +| Total record count | datastore row count; empty when the resource has no table | +| Resource update frequency | resource `update_frequency` | +| Temporal coverage from / to | resource `temporal_coverage`, split on `/` | +| Spatial coverage | resource `spatial_coverage` | + +The dataset is read with CKAN `package_show`. **json** is one object keyed by +the labels above (`Tags` is a list); **csv** is a header row and one value +row, with tags comma-joined. Named `_metadata.`. + +--- + ## `GET /datastore/api/v2/dump/query` Download the result of a **SQL `SELECT`** as a single file — filtered diff --git a/CLAUDE.md b/CLAUDE.md index a5deb81..b06ffaf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -433,6 +433,8 @@ Each endpoint takes a single `ContextDep`. The handler calls `context.authorize( | GET | `/datastore/api/v2/dump/query` | **implemented** | `sql=`, `format=csv\|gzip\|ndjson\|parquet` | 302 → GCS *or* streaming body (see §5.3) | | GET | `/datastore/api/v2/datastore_info` | **implemented** | `DatastoreInfoRequest` | `DatastoreInfoResponse` | | GET | `/datastore/api/v2/dump/{resource_id}` | **implemented** | `format=csv\|ndjson\|parquet` | 302 → GCS *or* streaming body (see §5.3) | +| GET | `/datastore/api/v2/dump/{resource_id}/dictionary` | **implemented** | `format=csv\|json` | 200 file — the resource's schema (`services/metadata.py`) | +| GET | `/datastore/api/v2/dump/{resource_id}/metadata` | **implemented** (`AUTH_TYPE=ckan`) | `format=csv\|json` | 200 file — resource + dataset metadata via CKAN `package_show` | The BigQuery engine is wired end-to-end: DDL, MERGE-based upsert, DML delete, parameterised search, native table-level metadata (the Frictionless schema + unique_key are JSON-encoded into the table's own `description` OPTION) for the schema round-trip, a row-count fast path via `INFORMATION_SCHEMA.TABLE_STORAGE`, and `EXPORT DATA`-backed dump with `table.modified`-keyed GCS caching. The DuckLake engine is the next concrete adapter — see §7. diff --git a/datastore/analytics.py b/datastore/analytics.py index d0fd4b6..33e18b7 100644 --- a/datastore/analytics.py +++ b/datastore/analytics.py @@ -62,6 +62,9 @@ def action_name(path: str) -> str | None: if path == DUMP_QUERY_PATH: return "datastore_dump_query" if path.startswith(DUMP_PATH_PREFIX): + document = path[len(DUMP_PATH_PREFIX) :].strip("/").split("/")[1:] + if document in (["dictionary"], ["metadata"]): + return f"datastore_dump_{document[0]}" return "datastore_dump" if path.startswith(ACTION_PREFIX): name = path[len(ACTION_PREFIX) :].strip("/").split("/", 1)[0] diff --git a/datastore/api/endpoints/dump.py b/datastore/api/endpoints/dump.py index fcbd6c9..67a4fda 100644 --- a/datastore/api/endpoints/dump.py +++ b/datastore/api/endpoints/dump.py @@ -1,4 +1,6 @@ -"""Download endpoints: `/{resource_id}` + `/query`. +"""Download endpoints: `/{resource_id}` + `/query`, +plus a resource's data dictionary and metadata at +`/{resource_id}/dictionary` and `…/metadata`. csv / gzip / ndjson shards are composed into one GCS object, so those always redirect — the server never touches the bytes. Parquet can't be @@ -13,16 +15,29 @@ from fastapi import APIRouter, Query from starlette.requests import Request -from starlette.responses import RedirectResponse, StreamingResponse +from starlette.responses import RedirectResponse, Response, StreamingResponse from datastore.api.context import Context from datastore.api.responses import ERROR_RESPONSES -from datastore.core.constants import DUMP_EXTENSIONS, DUMP_MEDIA_TYPES, DumpFormat +from datastore.core.constants import ( + DOCUMENT_MEDIA_TYPES, + DUMP_EXTENSIONS, + DUMP_MEDIA_TYPES, + DocumentFormat, + DumpFormat, +) from datastore.core.exceptions import ServerError from datastore.core.helper import content_disposition, dump_filename_base from datastore.infrastructure.engines import get_datastore_engine from datastore.schemas.request import DatastoreDumpSQLRequest from datastore.schemas.validators import ResourceId +from datastore.services.metadata import ( + dictionary_csv, + dictionary_document, + json_bytes, + metadata_csv, + metadata_record, +) from datastore.services.read import dump_sql_datastore from datastore.services.streaming import zip_archive_writer @@ -163,3 +178,80 @@ async def dump( urls = await engine.dump(resource_id, fmt, filename_base=filename_base) return download_response(request, urls, fmt, filename_base) + + +# What a dictionary / metadata download returns, for OpenAPI: the file +# itself, served by this API (no redirect — the documents are small). +_DOCUMENT_RESPONSES: dict[int | str, dict[str, Any]] = { + 200: { + "description": ( + "The file, named by `Content-Disposition`. Content type follows " + "`format`: " + + ", ".join(f"`{f}` → `{m}`" for f, m in DOCUMENT_MEDIA_TYPES.items()) + + "." + ), + "content": {media: {} for media in DOCUMENT_MEDIA_TYPES.values()}, + }, +} + + +def document_response( + body: bytes, fmt: DocumentFormat, authorized: dict[str, Any], resource_id: str, suffix: str +) -> Response: + """Serve an in-memory document, named like the resource's dump + (`_.`).""" + base = dump_filename_base(authorized.get("resource"), fmt) or resource_id + return Response( + content=body, + media_type=DOCUMENT_MEDIA_TYPES[fmt], + headers={"Content-Disposition": content_disposition(f"{base}_{suffix}.{fmt}")}, + ) + + +@router.get( + "/{resource_id}/dictionary", + summary="Download a resource's data dictionary", + response_class=Response, + responses=_DOCUMENT_RESPONSES, +) +async def dump_dictionary( + context: Context, + resource_id: ResourceId, + fmt: Annotated[DocumentFormat, Query(alias="format")] = "csv", +): + """Download the resource's table schema (data dictionary). + + `json` is the Frictionless Table Schema as `datastore_info` returns it. + `csv` has one row per column and one column per field property + (`name`, `type`, `title`, `description`, then any others); nested + values such as `constraints` are written as JSON in their cell.""" + authorized = await context.authorize(resource_id=resource_id, permission="read") + schema = await dictionary_document( + context, {"resource_id": resource_id, "resource": authorized.get("resource")} + ) + body = dictionary_csv(schema) if fmt == "csv" else json_bytes(schema) + return document_response(body, fmt, authorized, resource_id, "data_dictionary") + + +@router.get( + "/{resource_id}/metadata", + summary="Download a resource's and its dataset's metadata", + response_class=Response, + responses=_DOCUMENT_RESPONSES, +) +async def dump_metadata( + context: Context, + resource_id: ResourceId, + fmt: Annotated[DocumentFormat, Query(alias="format")] = "csv", +): + """Download the resource's metadata alongside its dataset's: names, + ids, description, organization, update frequencies, timestamps, + licence, tags, format, download URL, record count, temporal and + spatial coverage. + + `json` is one object; `csv` is a header row and one value row, with + tags comma-joined. Requires `AUTH_TYPE=ckan`.""" + authorized = await context.authorize(resource_id=resource_id, permission="read") + record = await metadata_record(context, resource_id, authorized) + body = metadata_csv(record) if fmt == "csv" else json_bytes(record) + return document_response(body, fmt, authorized, resource_id, "metadata") diff --git a/datastore/core/constants.py b/datastore/core/constants.py index 6dc402a..471aa5d 100644 --- a/datastore/core/constants.py +++ b/datastore/core/constants.py @@ -54,6 +54,15 @@ "parquet": "application/vnd.apache.parquet", } +# Formats for the data dictionary + metadata downloads +# (`/{rid}/dictionary`, `/{rid}/metadata`). Small +# documents built in memory, so the API serves the bytes itself. +DocumentFormat = Literal["csv", "json"] +DOCUMENT_MEDIA_TYPES: dict[str, str] = { + "csv": "text/csv; charset=utf-8", + "json": "application/json", +} + POSTGRES_TYPES: dict[str, str] = { # integer "int2": "int2", diff --git a/datastore/core/helper.py b/datastore/core/helper.py index 1a74074..e3d91a9 100644 --- a/datastore/core/helper.py +++ b/datastore/core/helper.py @@ -3,9 +3,14 @@ import os import re import unicodedata -from typing import Any +from typing import TYPE_CHECKING, Any from urllib.parse import quote, unquote, urlparse +from datastore.core.constants import DUMP_PREFIX + +if TYPE_CHECKING: + from datastore.core.config import Config + def parse_authorization_header(authorization: str | None) -> str | None: """Return the bare token from an `Authorization` header value, or None. @@ -84,3 +89,17 @@ def content_disposition(filename: str) -> str: fallback = fallback.replace('"', "").replace("\\", "") or "download" return f"attachment; filename=\"{fallback}\"; filename*=UTF-8''{quote(filename, safe='')}" return f'attachment; filename="{filename}"' + + +def dump_url(config: Config, resource_id: str) -> str: + """Absolute download URL for `resource_id` on this service. + + Base is `API_URL` when set, else `CKAN_URL`. The fallback is safe on + the CKAN paths that call this: under `AUTH_TYPE=ckan` a config + validator already guarantees `CKAN_URL` is non-empty — so the URL is + never left host-less. It does assume the datastore is reachable on + CKAN's host, which is true for the deployments this serves; a split + host needs `API_URL` set explicitly. + """ + base = (config.API_URL or config.CKAN_URL).rstrip("/") + return f"{base}{DUMP_PREFIX}/{resource_id}" diff --git a/datastore/infrastructure/ckan_client.py b/datastore/infrastructure/ckan_client.py index 5778729..b1d62b7 100644 --- a/datastore/infrastructure/ckan_client.py +++ b/datastore/infrastructure/ckan_client.py @@ -74,6 +74,11 @@ async def datastore_authorize( body["permission"] = permission return await self._post_action("datastore_authorize", body) + async def package_show(self, *, package_id: str) -> dict[str, Any]: + """`/api/3/action/package_show`. The full dataset dict, with its + resources, organization, tags and scheming extras.""" + return await self._post_action("package_show", {"id": package_id}) + async def resource_create(self, *, resource: dict[str, Any]) -> dict[str, Any]: """`/api/3/action/resource_create`. `resource` must include `package_id`.""" if not resource.get("package_id"): @@ -125,7 +130,6 @@ def _raise_for_status(action: str, response: httpx.Response) -> None: @staticmethod def _unwrap(action: str, response: httpx.Response) -> dict[str, Any]: """Parse the CKAN envelope; raise on `success=false` or bad shape.""" - print(response.status_code) try: payload = response.json() except ValueError as exc: diff --git a/datastore/services/metadata.py b/datastore/services/metadata.py new file mode 100644 index 0000000..6513352 --- /dev/null +++ b/datastore/services/metadata.py @@ -0,0 +1,152 @@ +"""Data dictionary + metadata documents for `/{rid}/…`. + +Both are small, so they're built in memory and handed to the endpoint as +bytes — no export pipeline, no GCS. CSV and JSON carry the same content; +CSV flattens anything nested into a single cell. +""" + +from __future__ import annotations + +import asyncio +import csv +import io +from typing import TYPE_CHECKING, Any + +import orjson + +from datastore.core.exceptions import NotFoundError, ValidationError +from datastore.core.helper import dump_url +from datastore.infrastructure.engines import get_datastore_engine +from datastore.services.read import info_datastore + +if TYPE_CHECKING: # type-only — no runtime import from api/ + from datastore.api.context import RequestContext + + +# Frictionless field properties that lead the dictionary CSV, in this +# order; any other property a field carries follows in first-seen order. +_DICTIONARY_LEADING_COLUMNS = ("name", "type", "title", "description") + + +async def dictionary_document(context: RequestContext, data_dict: dict[str, Any]) -> dict[str, Any]: + """The resource's Frictionless Table Schema — the same `schema` + `datastore_info` returns (CKAN's copy under `AUTH_TYPE=ckan`, else + the engine's).""" + result = await info_datastore(context, data_dict) + return result.schema + + +def dictionary_csv(schema: dict[str, Any]) -> bytes: + """One row per field, one column per field property.""" + fields = [f for f in schema.get("fields") or [] if isinstance(f, dict)] + columns = [c for c in _DICTIONARY_LEADING_COLUMNS if any(c in f for f in fields)] + for f in fields: + columns.extend(key for key in f if key not in columns) + return _csv_bytes(columns, fields) + + +async def metadata_record( + context: RequestContext, resource_id: str, authorized: dict[str, Any] +) -> dict[str, Any]: + """The resource's metadata plus its dataset's, one flat record keyed + by the labels the website shows. + + Reads the dataset through `package_show` rather than the dict + `datastore_authorize` returned: that one is cached by the auth + provider and isn't guaranteed to carry the organization, tags or + scheming extras. + """ + if context.config.AUTH_TYPE != "ckan" or context.ckan is None: + raise ValidationError("Metadata download requires AUTH_TYPE=ckan") + + authorized_resource = authorized.get("resource") or {} + package_id = authorized_resource.get("package_id") or (authorized.get("package") or {}).get( + "id" + ) + if not package_id: + raise NotFoundError(f"resource {resource_id!r} has no dataset") + + package = await context.ckan.package_show(package_id=str(package_id)) + resource = next( + (r for r in package.get("resources") or [] if r.get("id") == resource_id), + authorized_resource, + ) + temporal_from, temporal_to = _split_temporal_coverage(resource.get("temporal_coverage")) + + return { + "Dataset name": package.get("title"), + "Dataset id": package.get("id"), + "Dataset description": package.get("notes"), + "Data group": (package.get("organization") or {}).get("title"), + "Dataset update frequency": package.get("update_frequency"), + "Dataset created": package.get("metadata_created"), + "Dataset last updated": package.get("metadata_modified"), + "Licence": package.get("license_title") or package.get("license_id"), + "Tags": [t["name"] for t in package.get("tags") or [] if t.get("name")], + "Resource name": resource.get("name"), + "Resource id": resource_id, + "Format": resource.get("format"), + "Resource created": resource.get("created"), + # Same fallback the frontend shows as "updated", so the file and + # the page agree. + "Resource last updated": resource.get("metadata_modified") or resource.get("last_modified"), + "Download URL": dump_url(context.config, resource_id), + "Total record count": await _record_count(context, resource_id), + "Resource update frequency": resource.get("update_frequency"), + "Temporal coverage from": temporal_from, + "Temporal coverage to": temporal_to, + "Spatial coverage": resource.get("spatial_coverage"), + } + + +def metadata_csv(record: dict[str, Any]) -> bytes: + """A header row and one value row.""" + return _csv_bytes(list(record), [record]) + + +def json_bytes(document: Any) -> bytes: + return orjson.dumps(document, option=orjson.OPT_INDENT_2) + + +async def _record_count(context: RequestContext, resource_id: str) -> int | None: + """Row count from the engine; None when the resource has no table + yet — the rest of the metadata is still worth downloading.""" + engine = get_datastore_engine(context, mode="ro") + try: + info = await asyncio.to_thread(engine.info, resource_id=resource_id) + except NotFoundError: + return None + total = info.meta.get("total_records") + return total if isinstance(total, int) else None + + +def _split_temporal_coverage(value: Any) -> tuple[str | None, str | None]: + """`YYYY-MM-DD/YYYY-MM-DD` → (from, to). The validator enforces that + shape, but a legacy value without the `/` lands whole in `from`.""" + if not isinstance(value, str) or not value.strip(): + return None, None + start, sep, end = value.partition("/") + return start.strip() or None, (end.strip() or None) if sep else None + + +def _csv_bytes(columns: list[str], rows: list[dict[str, Any]]) -> bytes: + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\r\n") + writer.writerow(columns) + for row in rows: + writer.writerow([_csv_cell(row.get(c)) for c in columns]) + return buf.getvalue().encode("utf-8") + + +def _csv_cell(value: Any) -> str: + """Scalars as-is, a list of scalars comma-joined (tags), anything + else nested as compact JSON.""" + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (str, int, float)): + return str(value) + if isinstance(value, list) and all(isinstance(v, (str, int, float)) for v in value): + return ", ".join(str(v) for v in value) + return orjson.dumps(value).decode() diff --git a/datastore/services/write.py b/datastore/services/write.py index 42238e6..c014a7a 100644 --- a/datastore/services/write.py +++ b/datastore/services/write.py @@ -8,9 +8,8 @@ from frictionless import Resource, Schema from frictionless.exception import FrictionlessException -from datastore.core.config import Config -from datastore.core.constants import DUMP_PREFIX from datastore.core.exceptions import ValidationError +from datastore.core.helper import dump_url from datastore.infrastructure.engines import get_datastore_engine from datastore.schemas.responses import ( DatastoreCreateResponse, @@ -28,20 +27,6 @@ def _utc_now_iso() -> str: return datetime.now(timezone.utc).replace(tzinfo=None).isoformat() -def _dump_url(config: Config, resource_id: str) -> str: - """Absolute download URL for `resource_id` on this service. - - Base is `API_URL` when set, else `CKAN_URL`. The fallback is safe on - this path: it only runs under `AUTH_TYPE=ckan`, where a config - validator already guarantees `CKAN_URL` is non-empty — so the URL is - never left host-less. It does assume the datastore is reachable on - CKAN's host, which is true for the deployments this serves; a split - host needs `API_URL` set explicitly. - """ - base = (config.API_URL or config.CKAN_URL).rstrip("/") - return f"{base}{DUMP_PREFIX}/{resource_id}" - - def _validate_records( resource_id: str, schema: dict[str, Any], @@ -186,7 +171,7 @@ async def create_datastore( resource_id, schema=schema, url_type=url_type, - dump_url=_dump_url(context.config, resource_id), + dump_url=dump_url(context.config, resource_id), ) return DatastoreCreateResponse.Result( diff --git a/tests/conftest.py b/tests/conftest.py index e72a525..ecbdf00 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -103,6 +103,7 @@ def __init__(self) -> None: self.resources: dict[str, dict[str, Any]] = {} self.packages: dict[str, dict[str, Any]] = {} self.authorize_calls = 0 + self.package_show_calls = 0 self.deny_keys: set[str] = set() self._api_key: str | None = None @@ -152,6 +153,17 @@ async def datastore_authorize( "user": user, } + async def package_show(self, *, package_id: str) -> dict[str, Any]: + """Mirrors CKAN: the package with its resources embedded.""" + self.package_show_calls += 1 + if self._api_key and self._api_key in self.deny_keys: + raise AuthorizationError(f"key '{self._api_key}' is not allowed") + package = self.packages.get(package_id) + if package is None: + raise NotFoundError(f"package '{package_id}' not found") + resources = [r for r in self.resources.values() if r.get("package_id") == package_id] + return {**package, "resources": resources} + async def resource_create(self, *, resource: dict[str, Any]) -> dict[str, Any]: self._guard() package_id = str(resource.get("package_id") or "") diff --git a/tests/test_datastore_dump_documents.py b/tests/test_datastore_dump_documents.py new file mode 100644 index 0000000..46bc754 --- /dev/null +++ b/tests/test_datastore_dump_documents.py @@ -0,0 +1,258 @@ +"""Tests for `GET /{resource_id}/dictionary` and `…/metadata`. + +Both documents are built in memory and served as a 200 file. The +dictionary is the resource's schema (CKAN's copy under `AUTH_TYPE=ckan`); +the metadata joins the resource with its dataset from `package_show` +plus the engine's row count (0 from the placeholder BigQuery engine). +""" + +from __future__ import annotations + +import csv +import io +import json + +import pytest +from datastore.analytics import action_name +from datastore.core.constants import DUMP_PREFIX +from datastore.services.metadata import _csv_cell, _split_temporal_coverage +from fastapi.testclient import TestClient + +from tests.conftest import FakeCKAN + +RESOURCE_ID = "balancing_auction_results_2025" +DICTIONARY_URL = f"{DUMP_PREFIX}/{RESOURCE_ID}/dictionary" +METADATA_URL = f"{DUMP_PREFIX}/{RESOURCE_ID}/metadata" + +SCHEMA = { + "fields": [ + { + "name": "auction_id", + "type": "integer", + "title": "Auction ID", + "description": "Unique auction identifier.", + "constraints": {"required": True}, + }, + {"name": "product_code", "type": "string", "unit": "N/A"}, + ], + "primaryKey": ["auction_id"], +} + + +@pytest.fixture +def seeded(fake_ckan: FakeCKAN) -> FakeCKAN: + """A resource + dataset carrying every field the metadata reads.""" + fake_ckan.resources[RESOURCE_ID].update( + schema=SCHEMA, + format="CSV", + created="2025-01-01T09:00:00", + last_modified="2025-03-01T09:00:00", + update_frequency="daily", + temporal_coverage="2025-01-01/2025-12-31", + spatial_coverage="GB", + download_filename="Auction Results", + ) + fake_ckan.packages["pkg-balancing-2025"].update( + notes="Balancing auctions.", + organization={"name": "neso", "title": "NESO"}, + update_frequency="monthly", + metadata_created="2024-12-01T09:00:00", + metadata_modified="2025-02-01T09:00:00", + license_title="NESO Open Data Licence", + tags=[{"name": "balancing"}, {"name": "auctions"}], + ) + return fake_ckan + + +def _csv_rows(body: bytes) -> list[list[str]]: + return list(csv.reader(io.StringIO(body.decode("utf-8")))) + + +# --- dictionary ------------------------------------------------------------- + + +def test_dictionary_csv_has_a_row_per_field(client: TestClient, seeded: FakeCKAN) -> None: + response = client.get(DICTIONARY_URL) + + assert response.status_code == 200 + assert response.headers["content-type"] == "text/csv; charset=utf-8" + assert response.headers["content-disposition"] == ( + 'attachment; filename="Auction Results_data_dictionary.csv"' + ) + header, *rows = _csv_rows(response.content) + assert header == ["name", "type", "title", "description", "constraints", "unit"] + assert rows == [ + [ + "auction_id", + "integer", + "Auction ID", + "Unique auction identifier.", + '{"required":true}', + "", + ], + ["product_code", "string", "", "", "", "N/A"], + ] + + +def test_dictionary_json_is_the_schema(client: TestClient, seeded: FakeCKAN) -> None: + response = client.get(DICTIONARY_URL, params={"format": "json"}) + + assert response.status_code == 200 + assert response.headers["content-type"] == "application/json" + assert response.headers["content-disposition"].endswith('_data_dictionary.json"') + assert response.json() == SCHEMA + + +def test_dictionary_without_schema_is_header_only(client: TestClient) -> None: + """Placeholder engine + no CKAN schema → no fields, still a valid file.""" + response = client.get(DICTIONARY_URL) + + assert response.status_code == 200 + assert response.content == b"\r\n" + + +# --- metadata --------------------------------------------------------------- + + +def test_metadata_json_carries_every_field(client: TestClient, seeded: FakeCKAN) -> None: + response = client.get(METADATA_URL, params={"format": "json"}) + + assert response.status_code == 200 + assert response.headers["content-type"] == "application/json" + assert response.headers["content-disposition"] == ( + 'attachment; filename="Auction Results_metadata.json"' + ) + assert response.json() == { + "Dataset name": "Balancing Market 2025", + "Dataset id": "pkg-balancing-2025", + "Dataset description": "Balancing auctions.", + "Data group": "NESO", + "Dataset update frequency": "monthly", + "Dataset created": "2024-12-01T09:00:00", + "Dataset last updated": "2025-02-01T09:00:00", + "Licence": "NESO Open Data Licence", + "Tags": ["balancing", "auctions"], + "Resource name": "balancing-auction-results-2025", + "Resource id": RESOURCE_ID, + "Format": "CSV", + "Resource created": "2025-01-01T09:00:00", + # No `metadata_modified` on the resource → falls back to `last_modified`. + "Resource last updated": "2025-03-01T09:00:00", + "Download URL": f"https://example.com{DUMP_PREFIX}/{RESOURCE_ID}", + "Total record count": 0, + "Resource update frequency": "daily", + "Temporal coverage from": "2025-01-01", + "Temporal coverage to": "2025-12-31", + "Spatial coverage": "GB", + } + + +def test_metadata_prefers_resource_metadata_modified(client: TestClient, seeded: FakeCKAN) -> None: + seeded.resources[RESOURCE_ID]["metadata_modified"] = "2025-04-01T09:00:00" + + record = client.get(METADATA_URL, params={"format": "json"}).json() + + assert record["Resource last updated"] == "2025-04-01T09:00:00" + + +def test_metadata_csv_is_header_plus_one_row(client: TestClient, seeded: FakeCKAN) -> None: + response = client.get(METADATA_URL) + + assert response.status_code == 200 + assert response.headers["content-type"] == "text/csv; charset=utf-8" + header, *rows = _csv_rows(response.content) + assert len(rows) == 1 + row = dict(zip(header, rows[0], strict=True)) + assert header[0] == "Dataset name" + assert row["Tags"] == "balancing, auctions" + assert row["Data group"] == "NESO" + assert row["Total record count"] == "0" + + +def test_metadata_reads_the_dataset_through_package_show( + client: TestClient, seeded: FakeCKAN +) -> None: + """The dataset comes from `package_show`, not the (cached, possibly + partial) dict `datastore_authorize` returned.""" + record = client.get(METADATA_URL, params={"format": "json"}).json() + + assert seeded.package_show_calls == 1 + assert record["Dataset id"] == "pkg-balancing-2025" + + +def test_metadata_requires_ckan_auth(fake_ckan: FakeCKAN) -> None: + from datastore.api.context import get_auth_provider, get_ckan_client + from datastore.auth.anonymous import Provider as AnonymousProvider + from datastore.core.config import Config, get_config + from datastore.main import create_app + + app = create_app() + app.dependency_overrides[get_config] = lambda: Config(AUTH_TYPE="anonymous", CKAN_URL="") + app.dependency_overrides[get_ckan_client] = lambda: fake_ckan + app.dependency_overrides[get_auth_provider] = lambda: AnonymousProvider() + + with TestClient(app) as c: + response = c.get(METADATA_URL) + + assert response.status_code == 400 + assert "AUTH_TYPE=ckan" in response.json()["error"]["message"] + + +# --- shared ----------------------------------------------------------------- + + +@pytest.mark.parametrize("url", [DICTIONARY_URL, METADATA_URL]) +def test_unknown_format_is_rejected(client: TestClient, url: str) -> None: + response = client.get(url, params={"format": "parquet"}) + + assert response.status_code == 400 + assert response.json()["success"] is False + + +@pytest.mark.parametrize("document", ["dictionary", "metadata"]) +def test_unknown_resource_is_404(client: TestClient, document: str) -> None: + response = client.get(f"{DUMP_PREFIX}/no_such_resource/{document}") + + assert response.status_code == 404 + + +@pytest.mark.parametrize("document", ["dictionary", "metadata"]) +def test_denied_key_is_403(client: TestClient, fake_ckan: FakeCKAN, document: str) -> None: + fake_ckan.deny("test-token") + + response = client.get(f"{DUMP_PREFIX}/{RESOURCE_ID}/{document}") + + assert response.status_code == 403 + + +@pytest.mark.parametrize("document", ["dictionary", "metadata"]) +def test_analytics_names_each_document(document: str) -> None: + assert action_name(f"{DUMP_PREFIX}/res-1/{document}") == f"datastore_dump_{document}" + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("2025-01-01/2025-12-31", ("2025-01-01", "2025-12-31")), + ("2025-01-01", ("2025-01-01", None)), + ("", (None, None)), + (None, (None, None)), + ], +) +def test_split_temporal_coverage(value: str | None, expected: tuple[str | None, ...]) -> None: + assert _split_temporal_coverage(value) == expected + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (None, ""), + (True, "true"), + (3, "3"), + (["a", "b"], "a, b"), + ({"k": [1]}, json.dumps({"k": [1]}, separators=(",", ":"))), + ([{"k": 1}], '[{"k":1}]'), + ], +) +def test_csv_cell_flattens_nested_values(value: object, expected: str) -> None: + assert _csv_cell(value) == expected