feat(dump): download a resource's data dictionary and metadata

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 `<name>_data_dictionary.<format>`, where `<name>` 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 `<name>_metadata.<format>`.
+
+---
+
 ## `GET /datastore/api/v2/dump/query`
 
 Download the result of a **SQL `SELECT`** as a single file — filtered

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=<SELECT…>`, `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.
 

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]

datastore/api/endpoints/dump.py

@@ -1,4 +1,6 @@
-"""Download endpoints: `<DUMP_PREFIX>/{resource_id}` + `<DUMP_PREFIX>/query`.
+"""Download endpoints: `<DUMP_PREFIX>/{resource_id}` + `<DUMP_PREFIX>/query`,
+plus a resource's data dictionary and metadata at
+`<DUMP_PREFIX>/{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
+    (`<name>_<suffix>.<fmt>`)."""
+    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")

datastore/core/constants.py

@@ -54,6 +54,15 @@
     "parquet": "application/vnd.apache.parquet",
 }
 
+# Formats for the data dictionary + metadata downloads
+# (`<DUMP_PREFIX>/{rid}/dictionary`, `<DUMP_PREFIX>/{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",

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}"

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: