From 592e2dcb0616b3a6d737c1e7d000a785eeec1626 Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Tue, 22 Sep 2026 11:11:08 +0545 Subject: [PATCH 1/6] fix: keep backend detail out of public error responses Error envelopes returned the raw upstream exception, disclosing the query engine's name, the cloud region and internal job identifiers to anonymous callers. `APIError` gains a `detail` field: `message` is what the caller sees, `detail` is operator-only and reaches the log alone. Inverting the split this way makes the safe path the default - a new `ServerError("...")` without `detail` cannot leak. `str(exc)` still folds in the detail, so tracebacks stay useful. Every engine raise that interpolated an exception into `message` is converted, plus two of the same shape found by scanning for the pattern: the SQL-download dry run and the CKAN transport error, the latter carrying the internal cluster address. `_classify_bq_error` additionally maps BigQuery's caller-error reasons (`invalidQuery`, `invalid`, `notFound`) to a 400 rather than a 500 - an unknown column or a type-mismatched filter is the caller's mistake. The reason is read from `errors[0]`, not the `reason` attribute, which is None on a real `BadRequest`. 5xx now logs at ERROR with the detail and the original traceback; it was DEBUG, so in production the detail reached the client and nothing else. --- datastore/api/error_handlers.py | 12 +++- datastore/core/exceptions.py | 9 ++- datastore/infrastructure/ckan_client.py | 11 +++- .../engines/bigquery/backend.py | 61 ++++++++++++++++--- .../infrastructure/engines/bigquery/export.py | 36 +++++++++-- 5 files changed, 112 insertions(+), 17 deletions(-) diff --git a/datastore/api/error_handlers.py b/datastore/api/error_handlers.py index bf9046b..df13a99 100644 --- a/datastore/api/error_handlers.py +++ b/datastore/api/error_handlers.py @@ -42,13 +42,21 @@ def _group_errors(errors: list[dict[str, Any]]) -> dict[str, list[str]]: def register_exception_handlers(app: FastAPI) -> None: @app.exception_handler(APIError) async def _api_error(request: Request, exc: APIError) -> ORJSONResponse: - log.debug( - "APIError: %s -> %d (%s) at %s %s", + # 5xx means we are broken: log at ERROR with the operator-only + # `detail` and the original traceback. 4xx is the caller's doing + # and stays at DEBUG. `exc.message` is the only text that is ever + # serialised into the envelope. + is_server_fault = exc.status_code >= 500 + log.log( + logging.ERROR if is_server_fault else logging.DEBUG, + "APIError: %s -> %d (%s) at %s %s: %s", type(exc).__name__, exc.status_code, exc.type_label, request.method, request.url.path, + exc.detail or exc.message, + exc_info=exc.__cause__ if is_server_fault else None, ) return _error_response( request, diff --git a/datastore/core/exceptions.py b/datastore/core/exceptions.py index 266737c..d39652c 100644 --- a/datastore/core/exceptions.py +++ b/datastore/core/exceptions.py @@ -12,10 +12,17 @@ def __init__( message: str, *, fields: dict[str, list[str]] | None = None, + detail: str | None = None, ) -> None: - super().__init__(message) + # `str(exc)` carries the detail so tracebacks stay useful; the + # envelope is built from `.message` alone. + super().__init__(f"{message}: {detail}" if detail else message) + #: Sent to the caller. Keep it generic on 5xx - upstream text + #: names the engine, the cloud region and internal job ids. self.message = message self.fields = fields + #: Operator-only context. Logged, never serialised. + self.detail = detail class ValidationError(APIError): diff --git a/datastore/infrastructure/ckan_client.py b/datastore/infrastructure/ckan_client.py index c0e9173..5778729 100644 --- a/datastore/infrastructure/ckan_client.py +++ b/datastore/infrastructure/ckan_client.py @@ -27,6 +27,15 @@ } +def _ckan_transport_error(exc: Exception) -> ServerError: + """Wrap a CKAN transport failure. + + The httpx message embeds the upstream URL, which is an internal + cluster address; keep it in `detail` for the log. + """ + return ServerError("upstream request failed", detail=f"CKAN request failed: {exc}") + + class CKANClient: """Thin async wrapper around the CKAN action API. @@ -88,7 +97,7 @@ async def _post_action(self, action: str, body: dict[str, Any]) -> dict[str, Any try: response = await self._http.post(url, json=body, headers=headers) except httpx.HTTPError as exc: - raise ServerError(f"CKAN request failed: {exc}") from exc + raise _ckan_transport_error(exc) from exc self._raise_for_status(action, response) return self._unwrap(action, response) diff --git a/datastore/infrastructure/engines/bigquery/backend.py b/datastore/infrastructure/engines/bigquery/backend.py index 4d22727..d0f16da 100644 --- a/datastore/infrastructure/engines/bigquery/backend.py +++ b/datastore/infrastructure/engines/bigquery/backend.py @@ -30,6 +30,7 @@ from datastore.core.config import Config from datastore.core.exceptions import ( + APIError, NotFoundError, ServerError, ValidationError, @@ -66,6 +67,45 @@ log = logging.getLogger(__name__) + +# BigQuery reasons that mean "the caller sent something bad" rather than +# "we are broken": an unknown column, a filter value that will not parse +# as the column's type. Those are 400s. +_CLIENT_ERROR_REASONS = frozenset({"invalidQuery", "invalid", "notFound"}) + +_CLIENT_ERROR_MESSAGE = ( + "the query could not be executed; check column names, filter values " + "and types against the resource schema" +) + + +def _bq_error_reason(e: Exception) -> str: + """Pull BigQuery's machine-readable `reason`. + + `google.api_core` exposes it inside the `errors` list, *not* as an + attribute - `BadRequest.reason` is `None`. Reading the attribute + alone silently classifies every error as a server fault. + """ + for err in getattr(e, "errors", None) or (): + if isinstance(err, dict) and err.get("reason"): + return str(err["reason"]) + return "" + + +def _classify_bq_error(e: Exception, op: str, resource_id: str | None = None) -> APIError: + """Map a BigQuery exception to the right side of the 4xx/5xx line. + + Caller mistakes become `ValidationError`; everything else stays + `ServerError`. Either way the upstream text - engine name, region, + job id - rides in `detail`, which is logged and never serialised. + """ + where = f" for resource {resource_id!r}" if resource_id else "" + detail = f"BigQuery {op} failed{where}: {e}" + if _bq_error_reason(e) in _CLIENT_ERROR_REASONS: + return ValidationError(_CLIENT_ERROR_MESSAGE, detail=detail) + return ServerError("internal error", detail=detail) + + _HEALTHCHECK_TTL_SECONDS = 15.0 _HEALTHCHECK_TIMEOUT_SECONDS = 5.0 @@ -134,7 +174,8 @@ def _read_schema(self, resource_id: str) -> dict | None: return None except Exception as e: raise ServerError( - f"BigQuery tables.get failed for resource {resource_id!r}: {e}" + "internal error", + detail=f"BigQuery tables.get failed for resource {resource_id!r}: {e}", ) from e return table_to_schema(table) @@ -194,7 +235,10 @@ def _run_query( job.result() return job except Exception as e: - raise ServerError(f"BigQuery {op} failed for resource {resource_id!r}: {e}") from e + raise ServerError( + "internal error", + detail=f"BigQuery {op} failed for resource {resource_id!r}: {e}", + ) from e # ----- create helpers (DDL + records + branch orchestration) -------- def _create_table_sql(self, resource_id: str, schema: dict) -> str | None: @@ -611,7 +655,7 @@ def search( search_job = self.client.query(sql, job_config=job_config) row_iter = search_job.result() except Exception as e: - raise ServerError(f"BigQuery search failed for resource {resource_id!r}: {e}") from e + raise _classify_bq_error(e, "search", resource_id) from e total: int | None = None if include_total: @@ -622,9 +666,7 @@ def search( try: rows = list(count_job.result()) except Exception as e: - raise ServerError( - f"BigQuery search COUNT failed for resource {resource_id!r}: {e}" - ) from e + raise _classify_bq_error(e, "search COUNT", resource_id) from e total = int(rows[0]["n"]) if rows else 0 return SearchResult( @@ -674,7 +716,10 @@ def search_sql(self, sql: str, limit: int) -> SearchResult: dataset=self.config.BIGQUERY_DATASET, ) except Exception as e: - raise ServerError(f"failed to qualify table references in SQL: {e}") from e + raise ServerError( + "the SQL could not be prepared for execution", + detail=f"failed to qualify table references in SQL: {e}", + ) from e count_sql, count_params = self._search_sql_count_query(qualified_sql) @@ -699,7 +744,7 @@ def search_sql(self, sql: str, limit: int) -> SearchResult: ) row_iter = data_job.result() except Exception as e: - raise ServerError(f"BigQuery search_sql failed: {e}") from e + raise _classify_bq_error(e, "search_sql") from e total: int | None = None if count_job is not None: diff --git a/datastore/infrastructure/engines/bigquery/export.py b/datastore/infrastructure/engines/bigquery/export.py index 94e4f43..b2c654a 100644 --- a/datastore/infrastructure/engines/bigquery/export.py +++ b/datastore/infrastructure/engines/bigquery/export.py @@ -46,6 +46,20 @@ log = logging.getLogger(__name__) + +def _dry_run_error(e: Exception) -> ValidationError: + """Wrap a dry-run failure for the caller. + + The SQL is the caller's, so this is a 400 - but BigQuery's text names + the engine, the region and a job id, so only the generic line goes in + `message`. + """ + return ValidationError( + "the SQL is not valid for this datastore; check column names, " + "table names and syntax", + detail=f"sql failed BigQuery validation: {e}", + ) + # Per-format file extension + BigQuery EXPORT DATA `format` value. _FMT: dict[str, dict[str, str]] = { "csv": {"ext": "csv", "bq": "CSV"}, @@ -173,7 +187,10 @@ async def dump_sql( dataset=backend.config.BIGQUERY_DATASET, ) except Exception as e: - raise ServerError(f"failed to qualify table references in SQL: {e}") from e + raise ServerError( + "the SQL could not be prepared for execution", + detail=f"failed to qualify table references in SQL: {e}", + ) from e tables = {rid: await _get_table(backend, rid) for rid in resource_ids} @@ -201,7 +218,7 @@ async def build_export_sql(uri: str) -> tuple[str, bytes | None]: job_config=dry_cfg, ) except Exception as e: - raise ValidationError(f"sql failed BigQuery validation: {e}") from e + raise _dry_run_error(e) from e export_sql = _export_data_sql( uri, fmt, @@ -273,7 +290,10 @@ async def _prepare_download( log.info("BigQuery export cache MISS: %s attempt=%s", what, attempt) blobs = await _list_files_sorted(rw_gcs, attempt) if not blobs: - raise ServerError(f"BigQuery EXPORT DATA wrote no shards for {what}; check job logs.") + raise ServerError( + "the export produced no output", + detail=f"BigQuery EXPORT DATA wrote no shards for {what}; check job logs.", + ) blobs = await _compose_single_file(backend, rw_gcs, attempt, fmt, blobs, header_bytes) @@ -381,7 +401,10 @@ async def _run_export_job( f"Try {_FORMAT_HINT} for sharded multi-file downloads " "instead." ) from e - raise ServerError(f"BigQuery EXPORT DATA failed for {what}: {e}") from e + raise ServerError( + "the export failed", + detail=f"BigQuery EXPORT DATA failed for {what}: {e}", + ) from e log.debug( "BigQuery export: %s in %.2fs", @@ -673,4 +696,7 @@ async def _get_table(backend: Any, resource_id: str) -> Any: except NotFound as e: raise NotFoundError(f"resource {resource_id!r} is not declared; nothing to dump") from e except Exception as e: - raise ServerError(f"BigQuery get_table failed for resource {resource_id!r}: {e}") from e + raise ServerError( + "internal error", + detail=f"BigQuery get_table failed for resource {resource_id!r}: {e}", + ) from e From df1435a540fae9b11653591431889a8b10d8a1e7 Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Tue, 22 Sep 2026 11:11:08 +0545 Subject: [PATCH 2/6] fix: validate resource_id at the request boundary A `resource_id` containing a null byte was passed straight through to CKAN, which 500'd on it, and the API surfaced that as `Internal Error`. Malformed input is the caller's mistake and belongs in a 400 raised before any upstream call. Allow-list rather than block-list: `^[A-Za-z0-9_-]{1,200}$`. Blocking the null byte would only invite blocking the newline next, then the semicolon, and so on. The pattern covers both shapes actually in use - CKAN UUIDs and the table-style names `datastore_create` accepts and the docs' own examples use - so it rejects traversal, quotes, separators, control characters and over-length values without narrowing the API. A UUID-only rule, as the report suggested, would have broken documented behaviour. Applied to every `resource_id` field, to the `id` alias on info/delete (validating only one of the pair would leave a bypass), and to the `dump/{resource_id}` path parameter, which no request model covers. The rejection message states what is allowed rather than echoing the input, so control characters do not reach the response body or the logs. --- datastore/api/endpoints/dump.py | 3 ++- datastore/schemas/request.py | 16 ++++++++++------ datastore/schemas/validators.py | 26 ++++++++++++++++++++++++++ 3 files changed, 38 insertions(+), 7 deletions(-) diff --git a/datastore/api/endpoints/dump.py b/datastore/api/endpoints/dump.py index 47e4a73..bc16b7a 100644 --- a/datastore/api/endpoints/dump.py +++ b/datastore/api/endpoints/dump.py @@ -21,6 +21,7 @@ from datastore.core.exceptions import ServerError from datastore.infrastructure.engines import get_datastore_engine from datastore.schemas.request import DatastoreDumpSQLRequest +from datastore.schemas.validators import ResourceId from datastore.services.read import dump_sql_datastore from datastore.services.streaming import zip_archive_writer @@ -134,7 +135,7 @@ async def dump_sql( async def dump( request: Request, context: Context, - resource_id: str, + resource_id: ResourceId, fmt: Annotated[DumpFormat, Query(alias="format")] = "csv", ): """Download an entire resource; pick the format with `?format=`.""" diff --git a/datastore/schemas/request.py b/datastore/schemas/request.py index 51ac014..510a31e 100644 --- a/datastore/schemas/request.py +++ b/datastore/schemas/request.py @@ -15,6 +15,8 @@ from datastore.core.constants import DumpFormat from datastore.schemas.validators import ( FieldSpec, + OptionalResourceId, + ResourceId, StringOrList, fields_to_frictionless_schema, parse_sql_pagination, @@ -66,7 +68,7 @@ class DatastoreCreateRequest(BaseModel): }, ) - resource_id: str | None = Field( + resource_id: OptionalResourceId = Field( default=None, description="Target table name. Provide this **or** `resource`, not both.", ) @@ -188,7 +190,7 @@ class DatastoreUpsertRequest(BaseModel): }, ) - resource_id: str = Field( + resource_id: ResourceId = Field( description="Target table — must already exist (call `datastore_create` first)." ) records: list[dict[str, Any]] | None = Field(default=None, description="Rows to write.") @@ -231,7 +233,7 @@ class DatastoreSearchRequest(BaseModel): model_config = ConfigDict(extra="forbid") - resource_id: str = Field(description="Resource (table) to search.") + resource_id: ResourceId = Field(description="Resource (table) to search.") filters: str | None = Field( default=None, description=( @@ -456,8 +458,10 @@ class DatastoreInfoRequest(BaseModel): model_config = ConfigDict(extra="forbid") - resource_id: str | None = Field(default=None, description="Resource (table) to describe.") - id: str | None = Field( + resource_id: OptionalResourceId = Field( + default=None, description="Resource (table) to describe." + ) + id: OptionalResourceId = Field( default=None, description="CKAN alias for `resource_id`. Send exactly one." ) @@ -492,7 +496,7 @@ class DatastoreDeleteRequest(BaseModel): }, ) - resource_id: str | None = Field( + resource_id: OptionalResourceId = Field( default=None, description="Resource (table) to delete from / drop." ) id: str | None = Field( diff --git a/datastore/schemas/validators.py b/datastore/schemas/validators.py index e626c3a..c412792 100644 --- a/datastore/schemas/validators.py +++ b/datastore/schemas/validators.py @@ -31,6 +31,28 @@ def check_field_name(name: str) -> str: return name +# Resource (table) identifiers. CKAN hands out UUIDs; `datastore_create` +# also accepts table-style names, and the docs' own examples use them +# (`balancing_auction_results_2025`). Allow-list both shapes and reject +# everything else *before* the id reaches CKAN or BigQuery - a block-list +# of one control character, then the next, never finishes. +_RESOURCE_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,200}$") + + +def check_resource_id(value: Any) -> Any: + """Validate a resource id. Pass `None` through - the field's own + `| None` type decides whether absence is allowed.""" + if value is None: + return None + if not isinstance(value, str): + raise ValueError("resource_id must be a string") + if not _RESOURCE_ID_RE.match(value): + raise ValueError( + "resource_id must be valid" + ) + return value + + def to_list(value: Any) -> list[str] | None: """Coerce `None | str | list[str]` to `list[str] | None`. Equivalent to CKAN's `list_of_strings_or_string`. @@ -415,6 +437,10 @@ def rewrite_sql_offset( # `to_csv_list`) are invoked directly at the service boundary; they don't # need an Annotated wrapper because FastAPI's `Annotated[Model, Query()]` # only accepts scalar fields on the model — see DatastoreSearchRequest. +#: A validated resource (table) identifier. Use this instead of a bare +#: `str` on every field and path param that names a resource. +ResourceId = Annotated[str, BeforeValidator(check_resource_id)] +OptionalResourceId = Annotated[str | None, BeforeValidator(check_resource_id)] StringOrList = Annotated[list[str] | None, BeforeValidator(to_list)] PostgresType = Annotated[str | None, BeforeValidator(check_postgres_type)] From 185ce4a2230aca3a390db7c51fab90b28104c2a3 Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Tue, 22 Sep 2026 11:11:09 +0545 Subject: [PATCH 3/6] fix: sign download URLs with the read-only identity A V4 signed URL carries the signer's service-account email in `X-Goog-Credential`, where it is public. Signing with the rw account advertised the write identity on every download redirect, and issued a grant from an account that could also overwrite the object it points at. A blob signs through whichever client listed it, and the export path lists through rw, so `_signed_urls` re-binds each blob to the ro bucket before signing. Both the cache-hit and cache-miss paths converge on this one call, so neither can sign as rw. Only signing moves. The export, compose, publish and GC steps still run as rw, which they must. Requires an IAM change before deploy: the ro service account needs `iam.serviceAccountTokenCreator` on itself (IAM `signBlob` under workload identity) plus `storage.objects.get`. Signing will fail at runtime until that grant lands. Section 5.3's IAM table is updated to match - it previously documented signing as an rw operation. The service-account email cannot be removed from a signed URL; that is inherent to the signing scheme. What changes is that the account it names can only read. --- CLAUDE.md | 8 +++--- .../infrastructure/engines/bigquery/export.py | 25 +++++++++++++++---- 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index e5deb7a..80954e0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -465,7 +465,7 @@ dumps//// ← cache key, shared by all requests 4. **Compose** — csv/gzip/ndjson shards are stitched into ONE object server-side with GCS `compose` (≤32 sources per call, chained beyond that). csv and gzip compose a synthesized header member in first — plain bytes for csv, a gzip member for gzip — so the result carries exactly one header. (Verified against the real bucket: BigQuery's gzip objects carry **no** `Content-Encoding`, so compose byte-concats them and multi-member gzip decompresses as one file.) Parquet is never composable (footer + magic bytes) and stays as shards. Compose sources are the explicit shard list from **this attempt only** — never a fresh listing — so nothing foreign can be swept into the output. 5. **Publish** — write the zero-byte `/_SUCCESS`. This is the commit point: the attempt is unreadable before it, readable after, so no caller ever sees a half-written export. Written *after* the compose, so it can never become file content. 6. **Cleanup — in the background** (`_cleanup_in_background`; never on the response path): the compose source shards are deleted, and superseded attempts + revisions under `dumps///` are swept (`_delete_old_cache` — this request's own attempt is never touched, and the sweep is **always age-gated by the signed-URL expiry** so a sibling attempt that may still be exporting, or whose URLs are live, survives). -7. **Sign URLs** — V4 with `response-content-disposition: attachment; filename="."` (one file) or `_NN.` (multi-file parquet, 1-indexed). Signing is offloaded to a thread (IAM round-trip under workload identity). +7. **Sign URLs** — V4 with `response-content-disposition: attachment; filename="."` (one file) or `_NN.` (multi-file parquet, 1-indexed). Signing is offloaded to a thread (IAM round-trip under workload identity) and runs as the **ro** identity: the signer's SA email is public in the signed URL, so the account named there should be one that can only read. 8. **Return** (`download_response` in [api/endpoints/dump.py](datastore/api/endpoints/dump.py)): - 1 URL → `RedirectResponse(302)`. Bytes flow GCS → client; the server is **out of the byte path** for every format, and downloads are resumable. - N URLs (sharded parquet only) → `200` + **a streamed zip** of the parts (`zip_archive_writer` in [services/streaming.py](datastore/services/streaming.py)). The API fetches each signed URL over `app.state.http` and frames it into the archive a chunk at a time, so one export is always one file at one URL. Entries are `ZIP_STORED` (parquet is already compressed; deflating costs CPU for no size win) and `force_zip64=True` (member sizes aren't known up front). This is the **only** path where the server carries the bytes — no `Content-Length`, no range support, and a fetch failure mid-archive truncates a response that already returned 200. @@ -486,12 +486,12 @@ Required IAM. Dump follows a strict **ro for reading, rw for writing/updating** | `compose` (csv/ndjson) | RW GCS | Server-side concat into one object: reads each source's metadata (`storage.objects.get` — **`list` alone is not enough**) and creates the composite. | | `upload_from_string` (csv header member) | RW GCS | The one-row header composed in front of the header-less csv shards. | | `delete` (GC) | RW GCS | Writing/deleting objects. | -| `generate_signed_url` | RW GCS | Under workload identity this calls IAM `signBlob`, which typically only the rw SA holds via `iam.serviceAccountTokenCreator`. | +| `generate_signed_url` | **RO GCS** | A V4 URL carries the signer's SA email in `X-Goog-Credential`, where it is public, and the grant it issues is bounded by what that identity can do. Signing a GET with the rw SA advertises the write identity and issues a grant from an account that could also overwrite the object — so `_signed_urls` re-binds each blob to the ro bucket first. Requires `iam.serviceAccountTokenCreator` **on the ro SA** (IAM `signBlob` under workload identity) plus `storage.objects.get`. | Concrete perm sets: -- **RO SA** (`BIGQUERY_CREDENTIALS_RO`) — `bigquery.tables.get` + `storage.objects.list`. -- **RW SA** (`BIGQUERY_CREDENTIALS`) — `bigquery.jobs.create` + `bigquery.tables.export` + `bigquery.tables.getData` + `storage.objects.{create,get,list,delete}` + `iam.serviceAccountTokenCreator`. **`get` is required by `compose`** (it reads each source object); a role with only create/list/delete 403s on the compose step. +- **RO SA** (`BIGQUERY_CREDENTIALS_RO`) — `bigquery.tables.get` + `storage.objects.{list,get}` + `iam.serviceAccountTokenCreator` (on itself, for signing). +- **RW SA** (`BIGQUERY_CREDENTIALS`) — `bigquery.jobs.create` + `bigquery.tables.export` + `bigquery.tables.getData` + `storage.objects.{create,get,list,delete}`. **`get` is required by `compose`** (it reads each source object); a role with only create/list/delete 403s on the compose step. A single SA works if both perm sets land on the same identity — `BIGQUERY_CREDENTIALS_RO` empty falls through to ADC; same env var can drive both. `_build_bq_client` and `_build_storage_client` on the backend are deliberately small + stub-friendly so tests inject mocks without monkey-patching `google.cloud.*` globally. diff --git a/datastore/infrastructure/engines/bigquery/export.py b/datastore/infrastructure/engines/bigquery/export.py index b2c654a..fce1845 100644 --- a/datastore/infrastructure/engines/bigquery/export.py +++ b/datastore/infrastructure/engines/bigquery/export.py @@ -291,9 +291,9 @@ async def _prepare_download( blobs = await _list_files_sorted(rw_gcs, attempt) if not blobs: raise ServerError( - "the export produced no output", - detail=f"BigQuery EXPORT DATA wrote no shards for {what}; check job logs.", - ) + "the export produced no output", + detail=f"BigQuery EXPORT DATA wrote no shards for {what}; check job logs.", + ) blobs = await _compose_single_file(backend, rw_gcs, attempt, fmt, blobs, header_bytes) @@ -544,8 +544,13 @@ async def _signed_urls( ext: str, ) -> list[str]: """V4-sign each blob with an attachment filename (`.`, - or `_NN.` when there are several).""" + or `_NN.` when there are several). + """ expiry = _url_expiry(backend) + bucket_name = _bucket_of(blobs[0]) if blobs else None + ro_bucket = ( + backend._build_storage_client("ro").bucket(bucket_name) if bucket_name else None + ) def sign_all() -> list[str]: out: list[str] = [] @@ -555,8 +560,11 @@ def sign_all() -> list[str]: if len(blobs) == 1 else f"{filename_base}_{i + 1:02d}.{ext}" ) + # Re-bind to ro; fall back to the blob as listed if the + # bucket could not be resolved. + signer = ro_bucket.blob(blob.name) if ro_bucket is not None else blob out.append( - blob.generate_signed_url( + signer.generate_signed_url( version="v4", expiration=expiry, method="GET", @@ -568,6 +576,13 @@ def sign_all() -> list[str]: return await asyncio.to_thread(sign_all) +def _bucket_of(blob: Any) -> str | None: + """The bucket name a blob belongs to, or None if unresolvable.""" + bucket = getattr(blob, "bucket", None) + name = getattr(bucket, "name", None) + return name if isinstance(name, str) else None + + # ============================================================================ # 4. SQL / BYTES BUILDERS (pure) # ============================================================================ From d5e66b29db03d04ab02b7e17e632bbe7b00e9304 Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Tue, 22 Sep 2026 11:11:09 +0545 Subject: [PATCH 4/6] fix: emit the same timestamp precision in every dump format Parquet exported `*`, keeping BigQuery's native TIMESTAMP, while csv/ndjson cast to a whole-second ISO string. The same row therefore read back as `2026-09-11T07:08:35` from a csv download and `2026-09-11T07:08:35.908256` from a parquet one. Parquet now applies the same `format_select_column` cast as the other formats, keeping its JSON-to-string cast, which BigQuery still requires. `*` is retained when a table has neither timestamp nor JSON columns, so the common case keeps the short form. Affects both download endpoints, which share this select-list builder. Tradeoff: parquet now carries timestamps as strings rather than a native TIMESTAMP column, which is what matching csv exactly costs - csv can only carry text. If typed timestamps matter more than cross-format equality, the alternative is widening the csv format string to microseconds instead. --- .../infrastructure/engines/bigquery/export.py | 27 ++++++++++++++----- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/datastore/infrastructure/engines/bigquery/export.py b/datastore/infrastructure/engines/bigquery/export.py index fce1845..3082910 100644 --- a/datastore/infrastructure/engines/bigquery/export.py +++ b/datastore/infrastructure/engines/bigquery/export.py @@ -614,22 +614,35 @@ def _export_data_sql( def _export_select_list(schema: Any, fmt: str) -> str: - """SELECT column list: parquet casts JSON columns to strings (BQ - can't export native JSON to parquet); csv/ndjson use the same - `format_select_column` casts as `datastore_search`.""" + """SELECT column list for an export. + + Every format applies the same `format_select_column` casts as + `datastore_search`, so one row reads back identically whichever + format the caller picked. Parquet additionally casts JSON columns + to strings, because BigQuery cannot export native JSON to parquet. + + Parquet used to export `*`, which kept BigQuery's native TIMESTAMP + while csv/ndjson rendered a whole-second ISO string - the same row + came back as `...T07:08:35` in csv and `...T07:08:35.908256` in + parquet. `*` is still emitted when there is nothing to cast, which + keeps the common case free of a column list. + """ + fields = list(schema) if fmt == "parquet": - fields = list(schema) - if not any((f.field_type or "").upper() == "JSON" for f in fields): + if not any( + (f.field_type or "").upper() in ("JSON", "TIMESTAMP", "DATETIME") + for f in fields + ): return "*" return ", ".join( ( f"TO_JSON_STRING(`{f.name}`) AS `{f.name}`" if (f.field_type or "").upper() == "JSON" - else f"`{f.name}`" + else format_select_column(f.name, f.field_type) ) for f in fields ) - return ", ".join(format_select_column(f.name, f.field_type) for f in schema) + return ", ".join(format_select_column(f.name, f.field_type) for f in fields) def _csv_header_bytes(schema: Any, fmt: str) -> bytes: From 6f60256d711ac5b084e30b5e1b7701c6e4dcaa44 Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Tue, 22 Sep 2026 11:17:27 +0545 Subject: [PATCH 5/6] fix: return the CKAN envelope for unmatched routes and wrong methods A request to a path with no route, or with the wrong method, came back as FastAPI's default `{"detail": "Not Found"}` rather than the `{help, success, error:{__type, message}}` envelope every other error uses, so a client parsing the documented shape needed a special case. `fastapi.HTTPException` subclasses Starlette's, and it is Starlette's that the router raises for 404 and 405 - registering only the subclass never caught them. Registering the base class alongside it covers both without changing any existing behaviour. --- datastore/api/error_handlers.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/datastore/api/error_handlers.py b/datastore/api/error_handlers.py index df13a99..39fcf9b 100644 --- a/datastore/api/error_handlers.py +++ b/datastore/api/error_handlers.py @@ -6,6 +6,7 @@ from fastapi import FastAPI, HTTPException from fastapi.exceptions import RequestValidationError +from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.requests import Request from datastore.api.responses import ORJSONResponse, _error_response @@ -87,6 +88,12 @@ async def _validation(request: Request, exc: RequestValidationError) -> ORJSONRe fields=fields, ) + # Starlette's router raises *its* HTTPException for an unmatched route + # (404) or a wrong method (405). `fastapi.HTTPException` subclasses it, + # so registering the subclass alone left those two falling through to + # the default `{"detail": ...}` body - not the envelope every other + # error uses. Registering the base class covers both. + @app.exception_handler(StarletteHTTPException) @app.exception_handler(HTTPException) async def _http(request: Request, exc: HTTPException) -> ORJSONResponse: label = HTTP_STATUS_TO_TYPE_LABEL.get(exc.status_code, "Internal Error") From 3a5bd68e03462df37b664460354bb0fb41bd2f0a Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Tue, 22 Sep 2026 11:17:27 +0545 Subject: [PATCH 6/6] fix: reject a sort expression containing an empty entry `sort=", ,"` parsed to an empty list and was silently ignored, so a client that built the string wrong got unsorted rows and a 200 rather than being told. `"person_id,,first_name"` had the same problem mid- string, silently dropping nothing but hiding the typo. An empty term between commas is now a ValueError, which surfaces as a 400. A wholly blank `sort` still means "no sort" - that is absence rather than a malformed value, and remains accepted. --- .../infrastructure/engines/bigquery/search.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/datastore/infrastructure/engines/bigquery/search.py b/datastore/infrastructure/engines/bigquery/search.py index 959d1e0..bd21073 100644 --- a/datastore/infrastructure/engines/bigquery/search.py +++ b/datastore/infrastructure/engines/bigquery/search.py @@ -82,10 +82,20 @@ def parse_sort(sort_str: str, allowed: set[str]) -> list[tuple[str, str]]: name into the generated SQL (no parameter binding for identifiers). """ out: list[tuple[str, str]] = [] - for part in sort_str.split(","): + terms = sort_str.split(",") + # A wholly blank `sort` means "no sort" - absence, not a bad value. + # A blank *term* inside one (`"a,,b"`, `", ,"`) is a client mistake + # and used to be skipped silently, so a malformed sort returned + # unsorted rows with a 200. + if not sort_str.strip(): + return out + for part in terms: part = part.strip() if not part: - continue + raise ValueError( + f"sort {sort_str!r} has an empty entry; expected " + "'' or ' asc|desc' between commas" + ) tokens = part.split() if len(tokens) > 2: raise ValueError(