v1: per-deployment country allowlist and release stage, add DE on dev

src/quartz_api/cmd/main.py

@@ -349,6 +349,25 @@ def _create_v1_app(
 ) -> FastAPI:
     """Create and configure the v1 FastAPI sub-application."""
     v1_mod = importlib.import_module(service.__name__ + ".v1")
+    deployment = v1_mod.country_config
+    # Both sides come from the same env vars, so they only differ if a default was
+    # changed in one place and not the other.
+    scope = {
+        "v1_countries": (deployment.DEPLOYMENT_COUNTRIES or ""),
+        "v1_stage": deployment.DEPLOYMENT_STAGE,
+    }
+    for key, resolved in scope.items():
+        configured = conf.get_string(f"api.{key}", resolved)
+        if configured != resolved:
+            raise ValueError(
+                f"api.{key} is '{configured}' in server.conf but country_config "
+                f"resolved '{resolved}'. Keep their defaults in step.",
+            )
+    log.info(
+        "v1 serving countries %s at stage '%s'",
+        sorted(deployment.COUNTRIES),
+        deployment.DEPLOYMENT_STAGE,
+    )
 
     scalar_auth: dict = {}
     if auth_openapi_config:
@@ -392,7 +411,8 @@ def _create_v1_app(
     v1_app.add_exception_handler(grpc.aio.AioRpcError, _grpc_exception_handler)
     v1_app.add_middleware(SlowAPIMiddleware)
     v1_app.include_router(v1_mod.router)
-    v1_app.openapi = lambda: _custom_openapi(
+    # FastAPI's documented way to install a custom schema; ty sees a method overwritten.
+    v1_app.openapi = lambda: _custom_openapi(  # ty: ignore[invalid-assignment]
         v1_app, auth_openapi_config, contact=_V1_CONTACT,
     )
 
@@ -725,7 +745,8 @@ async def v1_scalar_docs(request: Request) -> HTMLResponse:
                       }
                     """,
         )
-        html = page.body.decode()
+        # Response.body is typed bytes | memoryview; HTMLResponse always gives bytes.
+        html = bytes(page.body).decode()
         if "</body>" not in html:
             return page
         return HTMLResponse(
@@ -984,7 +1005,8 @@ def redoc_html() -> FileResponse:
     server.description = description
 
     # Customize the OpenAPI schema (after auth config is resolved)
-    server.openapi = lambda: _custom_openapi(server)
+    # FastAPI's documented way to install a custom schema; ty sees a method overwritten.
+    server.openapi = lambda: _custom_openapi(server)  # ty: ignore[invalid-assignment]
 
     # Mount v1 as a sub-app (after auth config is resolved so v1 gets OAuth2 config)
     if "v1" in conf.get_string("api.routers").split(","):

src/quartz_api/cmd/server.conf

@@ -24,6 +24,15 @@ api {
   host_url = ${?HOST_URL} 
   workers = 1
   workers = ${?NUM_WORKERS}
+  // v1 deployment scope. service/v1/country_config.py reads these env vars directly,
+  // since the OpenAPI enums are built at import time. main.py refuses to start if the
+  // values here disagree with what it resolved, so change defaults in both places.
+  // Comma separated country codes to serve, e.g. "GB,NL". Empty serves all.
+  v1_countries = ""
+  v1_countries = ${?V1_COUNTRIES}
+  // "prod" hides catalogue entries tagged stage="dev"; "dev" shows them.
+  v1_stage = "prod"
+  v1_stage = ${?V1_STAGE}
 }
 
 // The backend to use for the service

src/quartz_api/cmd/test_main.py

@@ -1,5 +1,10 @@
 """Tests for the server assembly in main.py."""
 
+import json
+import os
+import subprocess
+import sys
+
 import grpc
 import pytest
 from fastapi import FastAPI
@@ -66,3 +71,86 @@ def test_grpc_error_repr_never_reaches_the_caller() -> None:
     resp = TestClient(_app_raising(grpc.StatusCode.NOT_FOUND)).get("/boom")
     assert "debug_error_string" not in resp.text
     assert "AioRpcError" not in resp.text
+
+
+# The v1 country filter runs when country_config is imported, and main.py builds the
+# server on import, so each case needs its own interpreter.
+_BOOT_PROBE = """
+import json
+from fastapi.testclient import TestClient
+from quartz_api.cmd.main import server
+client = TestClient(server)
+spec = client.get("/v1/openapi.json").json()
+print(json.dumps({
+    "enums": sorted({
+        tuple(sorted(p["schema"]["enum"]))
+        for path in spec["paths"].values()
+        for op in path.values()
+        for p in op.get("parameters", [])
+        if p["name"] == "country"
+    }),
+    "status": {
+        code: client.get(f"/v1/{code}/solar/region-types").status_code
+        for code in ("GB", "NL", "DE")
+    },
+}))
+"""
+
+
+def _boot(countries: str, stage: str) -> subprocess.CompletedProcess[str]:
+    env = {
+        **os.environ,
+        "ROUTERS": "v1",
+        "SOURCE": "dummydb",
+        "V1_COUNTRIES": countries,
+        "V1_STAGE": stage,
+    }
+    return subprocess.run(
+        [sys.executable, "-c", _BOOT_PROBE],
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=60,
+    )
+
+
+@pytest.mark.parametrize(
+    ("countries", "stage", "served"),
+    [
+        ("GB,NL", "prod", ["GB", "NL"]),
+        ("NL", "prod", ["NL"]),
+        # DE is dev-only, so allowlisting it on prod serves nothing extra.
+        ("NL,DE", "prod", ["NL"]),
+        ("GB,NL,DE", "dev", ["DE", "GB", "NL"]),
+    ],
+)
+def test_server_boots_serving_only_the_deployment_countries(
+    countries: str,
+    stage: str,
+    served: list[str],
+) -> None:
+    result = _boot(countries, stage)
+    assert result.returncode == 0, result.stderr
+    out = json.loads(result.stdout.strip().splitlines()[-1])
+    assert out["enums"] == [served]
+    assert out["status"] == {
+        code: 200 if code in served else 422 for code in ("GB", "NL", "DE")
+    }
+
+
+@pytest.mark.parametrize(
+    ("countries", "stage", "message"),
+    [
+        ("GB,GD", "prod", "V1_COUNTRIES has unknown codes ['GD']"),
+        ("GB", "production", "V1_STAGE must be one of"),
+    ],
+)
+def test_server_refuses_to_boot_on_bad_deployment_config(
+    countries: str,
+    stage: str,
+    message: str,
+) -> None:
+    result = _boot(countries, stage)
+    assert result.returncode != 0
+    assert message in result.stderr

src/quartz_api/conftest.py

@@ -0,0 +1,8 @@
+"""Test-wide setup that has to run before any quartz_api module is imported."""
+
+import os
+
+# country_config filters the v1 catalogue from these at import time. Tests assert on
+# every country and model, so a deployment scope in the shell must not leak in.
+os.environ.pop("V1_COUNTRIES", None)
+os.environ["V1_STAGE"] = "dev"

src/quartz_api/internal/middleware/audit.py

@@ -4,8 +4,9 @@
 from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING
 
-from fastapi import FastAPI, Request, Response
+from fastapi import Request, Response
 from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.types import ASGIApp
 
 if TYPE_CHECKING:
     from quartz_api.internal import models
@@ -14,9 +15,12 @@
 class RequestLoggerMiddleware(BaseHTTPMiddleware):
     """Middleware to log API requests to the database."""
 
-    def __init__(self, server: FastAPI) -> None:
-        """Initialize the middleware with the FastAPI server and database client."""
-        super().__init__(server)
+    def __init__(self, app: ASGIApp) -> None:
+        """Initialize the middleware.
+
+        `add_middleware` passes the wrapped ASGI app, not the FastAPI instance.
+        """
+        super().__init__(app)
 
     async def dispatch(
         self,

src/quartz_api/internal/middleware/sentry.py

@@ -4,8 +4,9 @@
 from collections.abc import Awaitable, Callable
 
 import sentry_sdk
-from fastapi import FastAPI, Request, Response
+from fastapi import Request, Response
 from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.types import ASGIApp
 
 from quartz_api.internal.middleware import auth
 from quartz_api.internal.middleware.auth import get_org_id_from_authdata
@@ -16,11 +17,14 @@ class SentryUserMiddleware(BaseHTTPMiddleware):
 
     def __init__(
         self,
-        server: FastAPI,
+        app: ASGIApp,
         auth_instance: auth.AuthClient | None,
     ) -> None:
-        """Initialize FastAPI server and auth instance."""
-        super().__init__(server)
+        """Initialize the middleware and auth instance.
+
+        `add_middleware` passes the wrapped ASGI app, not the FastAPI instance.
+        """
+        super().__init__(app)
         self.auth_instance = auth_instance
 
     async def dispatch(

src/quartz_api/internal/middleware/trace.py

@@ -9,8 +9,9 @@
 from typing import Any
 
 import grpc.aio
-from fastapi import FastAPI, Request, Response
+from fastapi import Request, Response
 from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.types import ASGIApp
 from typing_extensions import override
 
 CORR_HEADER = "X-Request-Id"
@@ -36,9 +37,12 @@ def set_trace_id(trace_id: str) -> None:
 class TracerMiddleware(BaseHTTPMiddleware):
     """Middleware to add tracing information to API requests."""
 
-    def __init__(self, server: FastAPI) -> None:
-        """Initialize the middleware with the FastAPI server and database client."""
-        super().__init__(server)
+    def __init__(self, app: ASGIApp) -> None:
+        """Initialize the middleware.
+
+        `add_middleware` passes the wrapped ASGI app, not the FastAPI instance.
+        """
+        super().__init__(app)
 
     async def dispatch(
         self,