feat: update openapi schema generation serializers/hooks

config/settings.py

@@ -2,8 +2,9 @@
 # Production settings
 import os
 import textwrap
+from collections.abc import Callable
 from pathlib import Path
-from typing import TypedDict
+from typing import Any, TypedDict
 
 
 class DatabaseSettings(TypedDict, total=False):
@@ -40,7 +41,7 @@ class TemplateSettings(TypedDict, total=False):
 class SpectacularSettings(TypedDict, total=False):
     TITLE: str
     DESCRIPTION: str
-    SORT_OPERATIONS: bool
+    SORT_OPERATIONS: bool | Callable[..., Any]
     SERVERS: list[dict[str, str]]
     EXTERNAL_DOCS: dict[str, str]
     VERSION: str
@@ -49,6 +50,7 @@ class SpectacularSettings(TypedDict, total=False):
     OAS_VERSION: str
     COMPONENT_SPLIT_REQUEST: bool
     TAGS: list[dict[str, str | dict[str, str]]]
+    POSTPROCESSING_HOOKS: list[str]
 
 
 BASE_DIR = Path(__file__).resolve().parent.parent
@@ -172,6 +174,35 @@ class SpectacularSettings(TypedDict, total=False):
 
 DEFAULT_AUTO_FIELD = "django.db.models.AutoField"
 
+
+def postprocesss_openapi_schema(result: dict[str, Any], generator: object, **kwargs: object) -> dict[str, Any]:
+    # root-level security marking all endpoints as public
+    result["security"] = []
+
+    page_size = REST_FRAMEWORK.get("PAGE_SIZE")
+    schemas = result.get("components", {}).get("schemas", {})
+    paths: dict[str, dict[str, Any]] = result.get("paths", {})
+    for path, path_item in paths.items():
+        for operation in path_item.values():
+            if isinstance(operation, dict) and operation.get("security") in ([{}], []):  # pyright: ignore[reportUnknownMemberType]
+                operation.pop("security", None)  # pyright: ignore[reportUnknownMemberType]
+
+        if page_size is None:
+            continue
+        response = path_item.get("get", {}).get("responses", {}).get("200", {})
+        schema = response.get("content", {}).get("application/json", {}).get("schema", {})
+        reference = schema.get("$ref", "")
+        if reference.startswith("#/components/schemas/"):
+            schema = schemas.get(reference.rsplit("/", 1)[-1], {})
+        properties = schema.get("properties", {})
+        if not {"count", "results", "next", "previous"}.issubset(properties):
+            continue
+        # Examples represent links from the second page.
+        for field, offset in (("next", 2 * page_size), ("previous", 0)):
+            properties[field]["example"] = f"https://pokeapi.co{path}?offset={offset}&limit={page_size}"
+    return result
+
+
 SPECTACULAR_SETTINGS: SpectacularSettings = {
     "TITLE": "PokéAPI",
     "DESCRIPTION": textwrap.dedent(
@@ -201,6 +232,9 @@ class SpectacularSettings(TypedDict, total=False):
     "AUTHENTICATION_WHITELIST": [],
     "OAS_VERSION": "3.1.0",
     "COMPONENT_SPLIT_REQUEST": True,
+    "POSTPROCESSING_HOOKS": [
+        "config.settings.postprocesss_openapi_schema",
+    ],
     "TAGS": [
         {
             "name": "pokemon",

data/v2/utils/media.py

@@ -5,7 +5,6 @@
 import urllib.error
 import urllib.request
 from collections.abc import Callable
-from email.utils import parsedate_to_datetime
 from pathlib import Path
 from typing import Any
 
@@ -36,29 +35,40 @@ def _load_manifest_from_path(path: Path) -> set[str] | None:
 
 
 def _fetch_manifest(url: str, cache_file: Path) -> set[str] | None:
+    etag_file = cache_file.with_suffix(cache_file.suffix + ".etag")
     try:
+        cached = None
+        headers: dict[str, str] = {}
         if cache_file.is_file():
-            try:
-                head_req = urllib.request.Request(url, method="HEAD")
-                with urllib.request.urlopen(head_req, timeout=10) as head_resp:
-                    last_mod_header = head_resp.headers.get("Last-Modified")
-                    if last_mod_header:
-                        remote_mtime = parsedate_to_datetime(last_mod_header).timestamp()
-                        if cache_file.stat().st_mtime >= remote_mtime:
-                            return _load_manifest_from_path(cache_file)
-            except (OSError, urllib.error.URLError, ValueError):
-                pass
-
-        with urllib.request.urlopen(url, timeout=30) as resp:
+            cached = _load_manifest_from_path(cache_file)
+            if cached is not None:
+                with contextlib.suppress(OSError, json.JSONDecodeError):
+                    metadata = json.loads(etag_file.read_text(encoding="utf-8"))
+                    if (
+                        isinstance(metadata, dict)
+                        and metadata.get("url") == url  # pyright: ignore[reportUnknownMemberType]
+                        and isinstance(metadata.get("etag"), str)  # pyright: ignore[reportUnknownMemberType]
+                    ):
+                        headers["If-None-Match"] = metadata["etag"]
+
+        request = urllib.request.Request(url, headers=headers)
+        try:
+            response = urllib.request.urlopen(request, timeout=30)
+        except urllib.error.HTTPError as e:
+            if e.code == 304 and cached is not None:
+                return cached
+            raise
+
+        with response as resp:
             raw_bytes: bytes = resp.read()
             data = json.loads(gzip.decompress(raw_bytes).decode("utf-8"))
 
             with contextlib.suppress(OSError):
+                etag_file.unlink(missing_ok=True)
                 cache_file.write_bytes(raw_bytes)
-                last_mod = resp.headers.get("Last-Modified")
-                if last_mod:
-                    mtime = parsedate_to_datetime(last_mod).timestamp()
-                    os.utime(cache_file, (mtime, mtime))
+                etag = resp.headers.get("ETag")
+                if etag:
+                    etag_file.write_text(json.dumps({"url": url, "etag": etag}), encoding="utf-8")
 
             return set(data)
     except (OSError, urllib.error.URLError, json.JSONDecodeError, gzip.BadGzipFile) as e: