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/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/api/error_handlers.py b/datastore/api/error_handlers.py index bf9046b..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 @@ -42,13 +43,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, @@ -79,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") 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..3082910 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", @@ -521,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] = [] @@ -532,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", @@ -545,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) # ============================================================================ @@ -576,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: @@ -673,4 +724,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 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( 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)]