feat(dump): vertical metadata CSV, system columns in data dictionary

datastore/api/endpoints/dump.py

@@ -223,8 +223,9 @@ async def dump_dictionary(
 ):
     """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
+    `json` is the Frictionless Table Schema as `datastore_info` returns it,
+    plus the system columns the data download carries (`_id` first,
+    `_updated_at` last when enabled). `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")
@@ -252,8 +253,8 @@ async def dump_metadata(
     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`."""
+    `json` is one object; `csv` has a `Field` and a `Value` column with
+    one row per metadata key, tags comma-joined. Requires `AUTH_TYPE=ckan`."""
     authorized = await context.authorize(resource_id=resource_id, permission="read")
     authorized["resource"] = await context.fresh_resource(resource_id, authorized.get("resource"))
     record = await metadata_record(context, resource_id, authorized)

datastore/services/metadata.py

@@ -14,6 +14,7 @@
 
 import orjson
 
+from datastore.core.constants import RESERVED_SYSTEM_COLUMN_NAMES
 from datastore.core.exceptions import NotFoundError, ValidationError
 from datastore.core.helper import dump_url
 from datastore.infrastructure.engines import get_datastore_engine
@@ -27,13 +28,29 @@
 # order; any other property a field carries follows in first-seen order.
 _DICTIONARY_LEADING_COLUMNS = ("name", "type", "title", "description")
 
+# Engine-managed columns every table carries but no user schema declares.
+# The dictionary lists them where downloads and `datastore_search` put
+# them: `_id` first, `_updated_at` last.
+_ID_FIELD = {
+    "name": "_id",
+    "type": "integer",
+    "title": "Row ID",
+    "description": "System generated. Row identifier assigned by the datastore.",
+}
+_UPDATED_AT_FIELD = {
+    "name": "_updated_at",
+    "type": "datetime",
+    "title": "Row last updated",
+    "description": "System generated. When the row was last inserted or changed (UTC).",
+}
+
 
 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)."""
+    the engine's), plus the system columns."""
     result = await info_datastore(context, data_dict)
-    return result.schema
+    return _with_system_fields(result.schema, context.config.INCLUDE_UPDATED_AT)
 
 
 def dictionary_csv(schema: dict[str, Any]) -> bytes:
@@ -100,8 +117,10 @@ async def metadata_record(
 
 
 def metadata_csv(record: dict[str, Any]) -> bytes:
-    """A header row and one value row."""
-    return _csv_bytes(list(record), [record])
+    """One row per metadata key: the label in the first column, its
+    value in the second."""
+    rows = [{"Field": key, "Value": value} for key, value in record.items()]
+    return _csv_bytes(["Field", "Value"], rows)
 
 
 def json_bytes(document: Any) -> bytes:
@@ -120,6 +139,18 @@ async def _record_count(context: RequestContext, resource_id: str) -> int | None
     return total if isinstance(total, int) else None
 
 
+def _with_system_fields(schema: dict[str, Any], include_updated_at: bool) -> dict[str, Any]:
+    """`schema` with `_id` prepended and `_updated_at` (when the engine
+    adds it) appended. A schema with no fields means no table to
+    describe, so it's returned as is."""
+    fields = [f for f in schema.get("fields") or [] if isinstance(f, dict)]
+    user = [f for f in fields if f.get("name") not in RESERVED_SYSTEM_COLUMN_NAMES]
+    if not user:
+        return schema
+    system_tail = [_UPDATED_AT_FIELD] if include_updated_at else []
+    return {**schema, "fields": [_ID_FIELD, *user, *system_tail]}
+
+
 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`."""

tests/test_datastore_dump_documents.py

@@ -15,7 +15,13 @@
 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 datastore.services.metadata import (
+    _ID_FIELD,
+    _UPDATED_AT_FIELD,
+    _csv_cell,
+    _split_temporal_coverage,
+    _with_system_fields,
+)
 from fastapi.testclient import TestClient
 
 from tests.conftest import FakeCKAN
@@ -82,6 +88,14 @@ def test_dictionary_csv_has_a_row_per_field(client: TestClient, seeded: FakeCKAN
     header, *rows = _csv_rows(response.content)
     assert header == ["name", "type", "title", "description", "constraints", "unit"]
     assert rows == [
+        [
+            "_id",
+            "integer",
+            "Row ID",
+            "System generated. Row identifier assigned by the datastore.",
+            "",
+            "",
+        ],
         [
             "auction_id",
             "integer",
@@ -91,16 +105,46 @@ def test_dictionary_csv_has_a_row_per_field(client: TestClient, seeded: FakeCKAN
             "",
         ],
         ["product_code", "string", "", "", "", "N/A"],
+        [
+            "_updated_at",
+            "datetime",
+            "Row last updated",
+            "System generated. When the row was last inserted or changed (UTC).",
+            "",
+            "",
+        ],
     ]
 
 
-def test_dictionary_json_is_the_schema(client: TestClient, seeded: FakeCKAN) -> None:
+def test_dictionary_json_is_the_schema_plus_system_columns(
+    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
+    assert response.json() == {
+        **SCHEMA,
+        "fields": [_ID_FIELD, *SCHEMA["fields"], _UPDATED_AT_FIELD],
+    }
+
+
+def test_system_fields_follow_include_updated_at() -> None:
+    assert _with_system_fields(SCHEMA, include_updated_at=False)["fields"] == [
+        _ID_FIELD,
+        *SCHEMA["fields"],
+    ]
+
+
+def test_system_fields_replace_declared_ones() -> None:
+    """A schema already naming a system column (e.g. inferred from an
+    unmanaged table) doesn't list it twice."""
+    schema = {"fields": [{"name": "_id", "type": "string"}, {"name": "a", "type": "string"}]}
+
+    names = [f["name"] for f in _with_system_fields(schema, include_updated_at=True)["fields"]]
+
+    assert names == ["_id", "a", "_updated_at"]
 
 
 def test_dictionary_without_schema_is_header_only(client: TestClient) -> None:
@@ -155,18 +199,20 @@ def test_metadata_prefers_resource_metadata_modified(client: TestClient, seeded:
     assert record["Resource last updated"] == "2025-04-01T09:00:00"
 
 
-def test_metadata_csv_is_header_plus_one_row(client: TestClient, seeded: FakeCKAN) -> None:
+def test_metadata_csv_is_one_row_per_key(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"
+    assert header == ["Field", "Value"]
+    assert all(len(r) == 2 for r in rows)
+    assert rows[0] == ["Dataset name", "Balancing Market 2025"]
+    values = dict(rows)
+    assert len(values) == len(rows) == 20
+    assert values["Tags"] == "balancing, auctions"
+    assert values["Data group"] == "NESO"
+    assert values["Total record count"] == "0"
 
 
 def test_metadata_reads_the_dataset_through_package_show(