Feat/legacy api support

.env.example

@@ -48,6 +48,7 @@ BIGQUERY_DATASET=
 BIGQUERY_CREDENTIALS=
 BIGQUERY_CREDENTIALS_RO=
 BIGQUERY_USE_QUERY_CACHE=true
+BIGQUERY_TRANSLATE_PG_SQL=true
 BIGQUERY_EXPORT_BUCKET=
 BIGQUERY_EXPORT_URL_EXPIRY_HOURS=1
 SQL_FUNCTIONS_ALLOW_FILE=

.gitignore

@@ -29,6 +29,11 @@ htmlcov/
 .idea/
 *.swp
 
+# Local-only testing / exploration (prod replication scripts, pulled
+# fixtures, production job logs). Everything here stays on the dev's
+# machine — never committed.
+/local_test/
+
 # OS
 .DS_Store
 Thumbs.db

CLAUDE.md

@@ -653,6 +653,8 @@ is easy to follow.
   "result": {
     "resource_id": "balancing_auction_results_2025",
     "fields": [
+      {"id": "_id",                        "type": "integer",  "info": {"title": "Record Identifier",      "...": "..."}},
+      {"id": "_updated_at",                "type": "datetime", "info": {"title": "Last Updated Timestamp", "...": "..."}},
       {"id": "auction_id",                 "type": "integer",  "info": {"title": "Auction ID", "...": "..."}},
       {"id": "product_code",               "type": "string",   "info": {"...": "..."}},
       {"id": "delivery_start",             "type": "datetime", "info": {"...": "..."}},
@@ -668,6 +670,13 @@ is easy to follow.
 }
 ```
 
+System columns are engine-managed — the request body must omit them
+(validation rejects `_id` / `_updated_at` in `fields`), and the service
+prepends them to the stored schema with service-provided `info`
+(title / description / comment / example / unit). The same `info` block
+rides through every read endpoint so clients see consistent column
+documentation.
+
 Optional response fields (omitted from the body when not requested):
 - `records` — echoes the input rows back when the request sets `include_records: true`.
 - `total_records` — total row count after the write, populated when `include_total: true`.
@@ -888,6 +897,9 @@ can confirm the table's new shape without a follow-up `datastore_info`:
 Returns the same field shape that was supplied to `datastore_create`, including
 the `info` data dictionary verbatim — clients can use this as a column-level
 metadata catalog (titles, descriptions, units, examples) without a side store.
+The two engine-managed system columns (`_id`, `_updated_at`) are prepended
+to `fields` with service-provided `info` so every read endpoint carries the
+same documentation for them.
 
 **Response**
 ```json
@@ -897,6 +909,28 @@ metadata catalog (titles, descriptions, units, examples) without a side store.
   "result": {
     "resource_id": "balancing_auction_results_2025",
     "fields": [
+      {
+        "id": "_id",
+        "type": "integer",
+        "info": {
+          "title": "Record Identifier",
+          "description": "System-generated unique identifier assigned to each record in the database. Used to uniquely identify a row within a resource table.",
+          "comment": "Automatically populated. Not publisher-maintained. Values are unique within a resource and may be used for record retrieval and updates.",
+          "example": "12345",
+          "unit": "N/A"
+        }
+      },
+      {
+        "id": "_updated_at",
+        "type": "datetime",
+        "info": {
+          "title": "Last Updated Timestamp",
+          "description": "System-generated timestamp indicating when the record was last created or modified in the database.",
+          "comment": "Automatically maintained. Useful for identifying recently changed records. Stored in UTC.",
+          "example": "2026-08-27T12:45:30",
+          "unit": "ISO 8601 UTC Timestamp"
+        }
+      },
       {
         "id": "auction_id",
         "type": "integer",

datastore/analytics.py

@@ -46,12 +46,26 @@
 
 from datastore.core.constants import API_PREFIX, DUMP_PREFIX
 
+# DEPRECATED legacy surface — this import, the two `LEGACY_*` constants
+# below, and the three `LEGACY_*` references downstream in this file
+# are the only legacy touch points in analytics. See
+# `datastore/legacy/__init__.py` for the full removal plan: when the
+# legacy URLs retire, delete lines marked `LEGACY ONLY` here plus the
+# single import above and analytics returns to the pure v2 shape.
+from datastore.legacy.constants import LEGACY_API_PREFIX, LEGACY_DUMP_PREFIX
+
 log = logging.getLogger(__name__)
 
 ACTION_PREFIX = f"{API_PREFIX}/"
 DUMP_PATH_PREFIX = f"{DUMP_PREFIX}/"
 DUMP_QUERY_PATH = f"{DUMP_PREFIX}/query"
 
+# LEGACY ONLY — CKAN-compatible surface. Recorded with the same action
+# names and event shape as v2 — downstream can split v2 vs legacy by
+# the `endpoint` path when it needs to.
+LEGACY_ACTION_PREFIX = f"{LEGACY_API_PREFIX}/"
+LEGACY_DUMP_PATH_PREFIX = f"{LEGACY_DUMP_PREFIX}/"
+
 # Served under the versioned prefix but not part of the API surface, so
 # they must not be recorded as action calls.
 DOCS_PATHS = frozenset({"docs", "redoc", "openapi.json", "static"})
@@ -71,6 +85,17 @@ def action_name(path: str) -> str | None:
         if not name or name in DOCS_PATHS:
             return None
         return name
+    # LEGACY ONLY — CKAN-compatible surface. Same action names as v2 so
+    # one event stream covers both API versions. The `endpoint` field
+    # preserves the raw path so a downstream filter can isolate legacy
+    # traffic. Delete this block when the legacy URLs retire.
+    if path.startswith(LEGACY_DUMP_PATH_PREFIX):
+        return "datastore_dump"
+    if path.startswith(LEGACY_ACTION_PREFIX):
+        name = path[len(LEGACY_ACTION_PREFIX) :].strip("/").split("/", 1)[0]
+        if not name:
+            return None
+        return name
     return None
 
 
@@ -172,7 +197,12 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         state: dict[str, Any] = scope.setdefault("state", {})
         status = 500  # what reaches the caller if the app dies before a response
         body = bytearray()
-        capture = scope["method"] == "POST" and scope["path"].startswith(ACTION_PREFIX)
+        # LEGACY ONLY: `LEGACY_ACTION_PREFIX` in the tuple keeps POST
+        # body capture working for `/api/3/action/*`. Drop that element
+        # when the legacy URLs retire.
+        capture = scope["method"] == "POST" and scope["path"].startswith(
+            (ACTION_PREFIX, LEGACY_ACTION_PREFIX)
+        )
 
         async def tee_receive() -> Any:
             message = await receive()
@@ -256,9 +286,16 @@ def _resource_ref(self, scope: Scope, body: bytes) -> str | None:
         those events rely on ``authorization_dict`` alone.
         """
         path: str = scope["path"]
-        if path.startswith(DUMP_PATH_PREFIX):
-            ref = path[len(DUMP_PATH_PREFIX) :].split("/", 1)[0]
-            return ref if ref and ref != "query" else None
+        # Both v2 (`/datastore/api/v2/dump/<rid>`) and legacy
+        # (`/datastore/dump/<rid>`) carry the resource id as the first
+        # path segment after the dump prefix. `query` is reserved on v2
+        # (SQL download endpoint) and not a resource id.
+        # LEGACY ONLY: drop `LEGACY_DUMP_PATH_PREFIX` from this tuple
+        # when the legacy URLs retire.
+        for prefix in (DUMP_PATH_PREFIX, LEGACY_DUMP_PATH_PREFIX):
+            if path.startswith(prefix):
+                ref = path[len(prefix) :].split("/", 1)[0]
+                return ref if ref and ref != "query" else None
 
         query: bytes = scope.get("query_string", b"")
         params = parse_qs(query.decode("latin-1"))

datastore/api/endpoints/datastore.py

@@ -184,13 +184,22 @@ async def datastore_search_sql(
     Accepts a single `sql` query parameter. To export the result as a
     file instead, use `GET <DUMP_PREFIX>/query`.
     """
+    # Collect each referenced resource's authorize payload — under
+    # AUTH_TYPE=ckan it carries the resource record (with its synced
+    # schema), which the service merges into the response so the output
+    # columns inherit CKAN-side data-dictionary edits by name.
+    resources: list[dict] = []
     for resource_id in params.resource_ids:
-        await context.authorize(resource_id=resource_id, permission="read")
+        auth_result = await context.authorize(resource_id=resource_id, permission="read")
+        resource = (auth_result or {}).get("resource")
+        if isinstance(resource, dict):
+            resources.append(resource)
 
     data_dict = params.model_dump() | {
         "function_names": params.function_names,
         "limit": params.limit,
         "offset": params.offset,
+        "resources": resources,
     }
 
     body_iter = await search_sql_datastore(

datastore/api/responses.py

@@ -34,7 +34,10 @@ def _orjson_default(obj: Any) -> Any:
     # values past ~15 significant digits round to the nearest IEEE-754
     # double (full-precision callers should CAST to STRING in SQL).
     if isinstance(obj, Decimal):
-        return float(obj)
+        f = float(obj)
+        if f.is_integer() and abs(f) < 2**63:
+            return int(f)
+        return f
     # `BYTES` columns come back as raw `bytes`; base64-encode so the
     # response stays UTF-8 and round-trippable.
     if isinstance(obj, bytes):

datastore/api/routes.py

@@ -5,6 +5,14 @@
 from datastore.api.endpoints import datastore, dump, health
 from datastore.core.constants import API_BASE_PREFIX, API_PREFIX, DUMP_PREFIX
 
+# DEPRECATED legacy CKAN surface. These two imports and the two
+# `legacy_endpoints` `include_router` calls below are the ONLY references
+# to `datastore.legacy` in the entire v2 codebase — remove them (plus
+# `datastore/legacy/` and `tests/test_legacy_api.py`) when the legacy URLs
+# retire. See `datastore/legacy/__init__.py` for the removal plan.
+from datastore.legacy import endpoints as legacy_endpoints
+from datastore.legacy.constants import LEGACY_API_PREFIX, LEGACY_DUMP_PREFIX
+
 api_router = APIRouter()
 # Unversioned: a probe URL is stable regardless of the action contract.
 api_router.include_router(health.probe_router, prefix=API_BASE_PREFIX)
@@ -14,3 +22,9 @@
 api_router.include_router(dump.router, prefix=DUMP_PREFIX)
 # Versioned: the datastore actions are the compatibility contract.
 api_router.include_router(datastore.router, prefix=API_PREFIX)
+
+# DEPRECATED — see `datastore/legacy/__init__.py`. These two mounts
+# expose read-only CKAN-compatible URLs so scripts in the wild keep
+# working while callers migrate to v2.
+api_router.include_router(legacy_endpoints.router, prefix=LEGACY_API_PREFIX)
+api_router.include_router(legacy_endpoints.dump_router, prefix=LEGACY_DUMP_PREFIX)

datastore/core/config.py

@@ -264,6 +264,19 @@ def _check_engine_available(cls, v: str) -> str:
             "force a fresh scan in tests."
         ),
     )
+    BIGQUERY_TRANSLATE_PG_SQL: bool = Field(
+        default=True,
+        description=(
+            "Translate Postgres-dialect SQL into BigQuery-dialect SQL before "
+            "running `datastore_search_sql` and `dump/query`. Covers the "
+            "pg-isms that broke customer scripts carried over from the "
+            "Postgres-backed CKAN datastore (double-quoted identifiers, "
+            "`::type` casts, `to_char`, `INTERVAL '10 days'`, "
+            "`CURRENT_DATE - 1`, …). Set False to disable the translator "
+            "per environment without a code change — SQL is then sent to "
+            "BigQuery as written."
+        ),
+    )
     BIGQUERY_EXPORT_BUCKET: str = Field(
         default="",
         description=("GCS bucket name (no `gs://` prefix) that `<DUMP_PREFIX>/<rid>` "),

datastore/core/constants.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Literal
+from typing import Any, Literal
 
 # Default for the `API_URL` setting — the public base URL used to render
 # absolute URLs in the OpenAPI examples. Deployments override it via env;
@@ -27,6 +27,9 @@
 API_VERSION = "v2"
 API_PREFIX = f"{API_BASE_PREFIX}/{API_VERSION}"
 DUMP_PREFIX = f"{API_PREFIX}/dump"
+# Note: legacy CKAN URL prefixes + their format tables live in
+# `datastore/legacy/constants.py`, kept out of this file so the whole
+# backwards-compat surface is one `rm -rf datastore/legacy/` away.
 # Download formats served by the export pipeline (`<DUMP_PREFIX>/…` and
 # `datastore_search_sql?download=…`). Lives here — not in `api/` — because
 # both the request schemas (pydantic layer) and the endpoints (starlette
@@ -156,10 +159,10 @@
 # Frictionless field type → canonical Postgres type. Used when deriving the
 # legacy `fields` shape from a Frictionless schema. Lossy in the other
 # direction (e.g., a Frictionless `integer` could have been any width); we
-# pick the widest/most-permissive Postgres type so the column accepts
-# anything the Frictionless type implies.
+# pick the type CKAN classic emits so the legacy wire shape matches —
+# `integer` → `int4` (both CKAN's user integers and `_id serial`).
 FRICTIONLESS_TO_POSTGRES: dict[str, str] = {
-    "integer": "int8",
+    "integer": "int4",
     "number": "numeric",
     "boolean": "bool",
     "string": "text",
@@ -207,3 +210,48 @@
         "_updated_at",
     }
 )
+
+
+# Service-provided documentation for the two engine-managed system
+# columns, shaped as full Frictionless field entries with an `info`
+# sub-dict so the read-side merger overlays them like any
+# publisher-supplied field. CKAN's sync strips these columns from a
+# resource's own schema, so without the seed `_id` / `_updated_at`
+# would arrive on every read as bare `{name, type}` entries with no
+# title or description.
+SYSTEM_FIELD_INFO: dict[str, dict[str, Any]] = {
+    "_id": {
+        "name": "_id",
+        "info": {
+            "title": "Record Identifier",
+            "description": (
+                "System-generated unique identifier assigned to each "
+                "record in the database. Used to uniquely identify a "
+                "row within a resource table."
+            ),
+            "comment": (
+                "Automatically populated. Not publisher-maintained. "
+                "Values are unique within a resource and may be used "
+                "for record retrieval and updates."
+            ),
+            "example": "12345",
+            "unit": "N/A",
+        },
+    },
+    "_updated_at": {
+        "name": "_updated_at",
+        "info": {
+            "title": "Last Updated Timestamp",
+            "description": (
+                "System-generated timestamp indicating when the record "
+                "was last created or modified in the database."
+            ),
+            "comment": (
+                "Automatically maintained. Useful for identifying "
+                "recently changed records. Stored in UTC."
+            ),
+            "example": "2026-08-27T12:45:30",
+            "unit": "ISO 8601 UTC Timestamp",
+        },
+    },
+}

datastore/infrastructure/engines/bigquery/backend.py

@@ -714,6 +714,7 @@ def search_sql(self, sql: str, limit: int) -> SearchResult:
                 sql,
                 project=self.config.BIGQUERY_PROJECT,
                 dataset=self.config.BIGQUERY_DATASET,
+                translate=self.config.BIGQUERY_TRANSLATE_PG_SQL,
             )
         except Exception as e:
             raise ServerError(

datastore/infrastructure/engines/bigquery/export.py

@@ -194,6 +194,7 @@ async def dump_sql(
             sql,
             project=backend.config.BIGQUERY_PROJECT,
             dataset=backend.config.BIGQUERY_DATASET,
+            translate=backend.config.BIGQUERY_TRANSLATE_PG_SQL,
         )
     except Exception as e:
         raise ServerError(