diff --git a/datastore/api/context.py b/datastore/api/context.py index f889187..d296cac 100644 --- a/datastore/api/context.py +++ b/datastore/api/context.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging from dataclasses import dataclass, field from typing import Annotated, Any @@ -12,9 +13,12 @@ from datastore.api.auth import Permission from datastore.auth.base import AuthProvider from datastore.core.config import Config, get_config +from datastore.core.exceptions import APIError from datastore.core.helper import parse_authorization_header from datastore.infrastructure.ckan_client import CKANClient +log = logging.getLogger(__name__) + ConfigDep = Annotated[Config, Depends(get_config)] # Declared as a security scheme (not a plain `Header`) so OpenAPI renders @@ -87,6 +91,24 @@ async def authorize( data_dict.pop("user", None) return data_dict + async def fresh_resource( + self, resource_id: str, cached: dict[str, Any] | None + ) -> dict[str, Any] | None: + """The live CKAN resource, for values that must not lag a replace. + + `authorize` returns the resource as it was when the decision was + cached (up to `AUTH_CACHE_TTL`), so a file replaced since then still + carries the old `url`. Falls back to `cached` when CKAN is + unreachable or standalone, so a CKAN blip never blocks a download. + """ + if self.ckan is None: + return cached + try: + return await self.ckan.resource_show(resource_id=resource_id) + except APIError: + log.warning("resource_show failed for %s; using cached resource", resource_id) + return cached + def get_context( request: Request, diff --git a/datastore/api/endpoints/dump.py b/datastore/api/endpoints/dump.py index 67a4fda..10617fa 100644 --- a/datastore/api/endpoints/dump.py +++ b/datastore/api/endpoints/dump.py @@ -128,15 +128,17 @@ async def dump_sql( """Download query result as a file, named `_filtered_` when the query reads a single resource that has a download name (see `dump`), else `query_`.""" - resources = [ - (await context.authorize(resource_id=resource_id, permission="read")).get("resource") + authorized = [ + await context.authorize(resource_id=resource_id, permission="read") for resource_id in params.resource_ids ] - name = ( - dump_filename_base(resources[0], DUMP_EXTENSIONS[params.format]) - if len(resources) == 1 - else None - ) + name = None + if len(authorized) == 1: + # Live resource: the cached one still holds the pre-replace url. + resource = await context.fresh_resource( + params.resource_ids[0], authorized[0].get("resource") + ) + name = dump_filename_base(resource, DUMP_EXTENSIONS[params.format]) filename_prefix = f"{name}_filtered" if name else None urls = await dump_sql_datastore( @@ -171,9 +173,9 @@ async def dump( resources, else the resource's `download_filename`; when neither is available, after the resource id.""" authorized = await context.authorize(resource_id=resource_id, permission="read") - filename_base = ( - dump_filename_base(authorized.get("resource"), DUMP_EXTENSIONS[fmt]) or resource_id - ) + # Live resource: the cached one still holds the pre-replace url. + resource = await context.fresh_resource(resource_id, authorized.get("resource")) + filename_base = dump_filename_base(resource, DUMP_EXTENSIONS[fmt]) or resource_id engine = get_datastore_engine(context, mode="ro") urls = await engine.dump(resource_id, fmt, filename_base=filename_base) @@ -226,6 +228,7 @@ async def dump_dictionary( (`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") + authorized["resource"] = await context.fresh_resource(resource_id, authorized.get("resource")) schema = await dictionary_document( context, {"resource_id": resource_id, "resource": authorized.get("resource")} ) @@ -252,6 +255,7 @@ async def dump_metadata( `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") + authorized["resource"] = await context.fresh_resource(resource_id, authorized.get("resource")) 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") diff --git a/datastore/infrastructure/ckan_client.py b/datastore/infrastructure/ckan_client.py index b1d62b7..9e1f0fb 100644 --- a/datastore/infrastructure/ckan_client.py +++ b/datastore/infrastructure/ckan_client.py @@ -74,6 +74,10 @@ async def datastore_authorize( body["permission"] = permission return await self._post_action("datastore_authorize", body) + async def resource_show(self, *, resource_id: str) -> dict[str, Any]: + """`/api/3/action/resource_show`. The live resource dict.""" + return await self._post_action("resource_show", {"id": resource_id}) + 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.""" diff --git a/tests/conftest.py b/tests/conftest.py index ecbdf00..2a85c58 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -39,7 +39,7 @@ import pytest # noqa: E402 from datastore.api.context import get_auth_provider, get_ckan_client # noqa: E402 from datastore.auth.ckan import Provider as CKANAuthProvider # noqa: E402 -from datastore.core.exceptions import AuthorizationError, NotFoundError # noqa: E402 +from datastore.core.exceptions import AuthorizationError, NotFoundError, ServerError # noqa: E402 from datastore.infrastructure.cache import InMemoryCache # noqa: E402 from datastore.main import create_app # noqa: E402 from fastapi.testclient import TestClient # noqa: E402 @@ -105,6 +105,10 @@ def __init__(self) -> None: self.authorize_calls = 0 self.package_show_calls = 0 self.deny_keys: set[str] = set() + # What `datastore_authorize` still reports after a resource changed, + # i.e. the entry the provider's TTL cache would hand back. + self.stale_resources: dict[str, dict[str, Any]] = {} + self.fail_resource_show = False self._api_key: str | None = None def add_resource(self, resource_id: str, **extra: Any) -> None: @@ -141,7 +145,8 @@ async def datastore_authorize( package = self.packages.get(pkg_id) if package is None: raise NotFoundError(f"package '{pkg_id}' not found") - return {"package": package, "resource": existing, "user": user} + resource = self.stale_resources.get(resource_id, existing) + return {"package": package, "resource": resource, "user": user} assert package_id is not None package = self.packages.get(package_id) @@ -153,6 +158,14 @@ async def datastore_authorize( "user": user, } + async def resource_show(self, *, resource_id: str) -> dict[str, Any]: + if self.fail_resource_show: + raise ServerError("CKAN returned 500 for resource_show") + existing = self.resources.get(resource_id) + if existing is None: + raise NotFoundError(f"resource '{resource_id}' not found") + return existing + async def package_show(self, *, package_id: str) -> dict[str, Any]: """Mirrors CKAN: the package with its resources embedded.""" self.package_show_calls += 1 diff --git a/tests/test_datastore_dump.py b/tests/test_datastore_dump.py index 78d13fa..6226a1f 100644 --- a/tests/test_datastore_dump.py +++ b/tests/test_datastore_dump.py @@ -181,6 +181,41 @@ def test_dump_of_upload_uses_uploaded_filename(client: TestClient, fake_ckan: Fa assert calls[0]["filename_base"] == "auction-results" +def test_dump_names_file_after_replaced_upload_despite_stale_auth( + client: TestClient, fake_ckan: FakeCKAN +) -> None: + """The auth check can still report the pre-replace url (TTL cache); the + dump must be named after the file CKAN holds now.""" + rid = "balancing_auction_results_2025" + old = { + **fake_ckan.resources[rid], + "url_type": "upload", + "url": "https://ckan/dataset/d/resource/r/download/first.csv", + } + fake_ckan.stale_resources[rid] = old + fake_ckan.resources[rid] = { + **old, + "url": "https://ckan/dataset/d/resource/r/download/second.csv", + } + calls: list[dict[str, Any]] = [] + with _patch_dump(["https://example/a.ndjson"], calls): + client.get(DUMP_URL, params={"format": "ndjson"}, follow_redirects=False) + assert calls[0]["filename_base"] == "second" + + +def test_dump_falls_back_to_cached_resource_when_ckan_fails( + client: TestClient, fake_ckan: FakeCKAN +) -> None: + rid = "balancing_auction_results_2025" + fake_ckan.resources[rid]["download_filename"] = "Cached Name" + fake_ckan.fail_resource_show = True + calls: list[dict[str, Any]] = [] + with _patch_dump(["https://example/a.csv"], calls): + response = client.get(DUMP_URL, follow_redirects=False) + assert response.status_code == 302 + assert calls[0]["filename_base"] == "Cached Name" + + def test_sharded_parquet_zip_uses_download_filename( client: TestClient, fake_ckan: FakeCKAN ) -> None: