From 5bf50c8af8727e77b5d5b7fc88fdecaeabaac5a2 Mon Sep 17 00:00:00 2001 From: Kelsey Smuczynski Date: Fri, 17 Jul 2026 23:15:42 -0600 Subject: [PATCH 1/7] feat(ogc): add auth gate for internal OGC mount pygeoapi is mounted via a raw Starlette Mount, so FastAPI's Depends() machinery never runs for it -- auth has to happen at the ASGI layer, in front of the mount, not as a route dependency. Added TokenInvalid and decode_token_payload() alongside the existing _get_token_payload so the same JWKS/jwt.decode logic can be reused from code with no FastAPI exception handler watching (raising HTTPException there would just be an unhandled exception, not a response). InternalOGCAuthMiddleware is a plain ASGI middleware class rather than BaseHTTPMiddleware (used elsewhere in this app for request logging and lazy admin init): BaseHTTPMiddleware buffers the full response body and breaks client-disconnect propagation, which matters here since this mount serves paginated GeoJSON up to max_items: 10000. Splits 401 (no/invalid token) from 403 (valid token, wrong group) per the acceptance criteria, rather than collapsing both into 401 like every other auth path in this app does today -- the underlying check is identical either way, so the split is a one-line branch, not meaningfully more code to maintain. --- core/dependencies.py | 6 ++ core/internal_ogc_auth.py | 115 ++++++++++++++++++++++++++++++++++++++ core/permissions.py | 32 +++++++++++ 3 files changed, 153 insertions(+) create mode 100644 core/internal_ogc_auth.py diff --git a/core/dependencies.py b/core/dependencies.py index eabcd009a..6372804a9 100644 --- a/core/dependencies.py +++ b/core/dependencies.py @@ -56,6 +56,12 @@ lexicon_editor_function = authenticated(permissions=["LexiconEditor"]) +# OGC-Internal Authentication/Permissions -------------------------------------- +# INTERNAL_OGC_GROUP ("OGCInternal") lives in core/permissions.py, not here -- +# it gates core/internal_ogc_auth.py's ASGI middleware in front of the +# /ogcapi-internal mount, which runs outside FastAPI's Depends() machinery. + + # Testing-Specific Authentication/Permissions ---------------------------------- no_permission_function = authenticated(permissions=["NoPermission"]) diff --git a/core/internal_ogc_auth.py b/core/internal_ogc_auth.py new file mode 100644 index 000000000..8edd767cf --- /dev/null +++ b/core/internal_ogc_auth.py @@ -0,0 +1,115 @@ +# =============================================================================== +# Copyright 2026 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +"""ASGI auth gate for the authenticated internal OGC mount (/ogcapi-internal). + +pygeoapi is mounted via a raw Starlette Mount (core/pygeoapi.py), so FastAPI's +Depends() machinery never runs for it -- gating has to happen at the ASGI +layer, in front of the mount. This is a plain ASGI middleware class rather +than @app.middleware("http")/BaseHTTPMiddleware (used elsewhere in this +codebase): BaseHTTPMiddleware buffers the full response body and interferes +with client-disconnect propagation, which matters here since +/ogcapi-internal serves paginated GeoJSON up to `max_items: 10000`. On the +success path this calls straight through with zero buffering. + +Kept separate from core/permissions.py to avoid a circular import with +core/pygeoapi.py. +""" + +import json +import os + +from starlette.types import ASGIApp, Receive, Scope, Send + +from core import permissions +from core.settings import settings + + +def _extract_bearer_token(scope: Scope) -> str | None: + headers = dict(scope.get("headers") or []) + authorization = headers.get(b"authorization") + if not authorization: + return None + scheme, _, param = authorization.decode("latin-1").partition(" ") + if scheme.lower() != "bearer" or not param: + return None + return param + + +async def _send_json(send: Send, status_code: int, detail: str) -> None: + body = json.dumps({"detail": detail}).encode("utf-8") + await send( + { + "type": "http.response.start", + "status": status_code, + "headers": [(b"content-type", b"application/json")], + } + ) + await send({"type": "http.response.body", "body": body}) + + +class InternalOGCAuthMiddleware: + """Gates every request under `mount_path` behind INTERNAL_OGC_GROUP + membership; requests to any other path pass straight through untouched. + + Registered via app.add_middleware(), which wraps the whole app -- the + path check below is what keeps this scoped to the internal mount only. + """ + + def __init__(self, app: ASGIApp, mount_path: str) -> None: + self.app = app + self.mount_path = mount_path + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http" or not scope["path"].startswith(self.mount_path): + await self.app(scope, receive, send) + return + + if int(os.environ.get("AUTHENTIK_DISABLE_AUTHENTICATION", 0)): + if settings.mode == "production": + # HTTPException(424) (what core.permissions.authenticated() + # raises for this same misconfiguration) means nothing from + # raw ASGI code -- send the response directly so a + # misconfigured production box degrades to "internal mount + # always 424s" rather than crashing the worker. + await _send_json( + send, + 424, + "Authentication is disabled in production mode. Set " + "AUTHENTIK_DISABLE_AUTHENTICATION=0 to enable authentication.", + ) + return + await self.app(scope, receive, send) + return + + token = _extract_bearer_token(scope) + if not token: + await _send_json(send, 401, "Unauthorized") + return + + try: + payload = permissions.decode_token_payload(token) + except permissions.TokenInvalid: + await _send_json(send, 401, "Could not validate credentials") + return + + if permissions.INTERNAL_OGC_GROUP not in payload.get("groups", []): + await _send_json(send, 403, "Forbidden") + return + + await self.app(scope, receive, send) + + +# ============= EOF ============================================= diff --git a/core/permissions.py b/core/permissions.py index 952e844f4..fec27d37f 100644 --- a/core/permissions.py +++ b/core/permissions.py @@ -153,4 +153,36 @@ def _get_token_payload(token: str = Depends(oauth2_scheme)): ) +class TokenInvalid(Exception): + """Raised by decode_token_payload() for any JWT verification failure. + + Not an HTTPException: this is called from raw ASGI middleware + (core/internal_ogc_auth.py), which has no FastAPI exception handler + watching, so raising HTTPException there would just be an unhandled + exception rather than the intended response. + """ + + +# Required Authentik group for the authenticated internal OGC mount +# (/ogcapi-internal). Not Depends()-shaped like the roles above -- see the +# cross-reference note in core/dependencies.py for why it still lives here. +INTERNAL_OGC_GROUP = "OGCInternal" + + +def decode_token_payload(token: str) -> dict: + """Same JWT verification as _get_token_payload (get_public_key/JWKS/ + jwt.decode), but raises TokenInvalid instead of HTTPException(401). + """ + try: + public_key = get_public_key(token) + return jwt.decode( + token, + public_key, + algorithms=ALGORITHMS, + audience=os.environ.get("AUTHENTIK_CLIENT_ID"), + ) + except (JWTError, HTTPException) as e: + raise TokenInvalid(str(e)) from e + + # ============= EOF ============================================= From d2e874943d9941f26fb74c60ac341e5da077fcc8 Mon Sep 17 00:00:00 2001 From: Kelsey Smuczynski Date: Fri, 17 Jul 2026 23:18:06 -0600 Subject: [PATCH 2/7] feat(ogc): add authenticated internal OGC mount Generalized _mount_path()/_pygeoapi_dir()/_write_config() to take an env var and default rather than hardcoding the public mount's values, so the internal mount gets the same traversal/character safety and sensitive-file handling for free instead of a copy-pasted second implementation. mount_pygeoapi_internal() gets its own guard flag and runtime dir so it can't silently no-op against the public mount's state, and a startup check that the two configured mount paths actually differ -- Starlette doesn't error on duplicate Mounts, it just routes to whichever registered first, which would leave the internal mount silently unreachable rather than failing loudly. Also added _assert_server_settings_match(): pygeoapi.api.API mutates process-wide globals (CHARSET, FORMAT_TYPES) during __init__, so whichever of the two mounts is built last wins for both. Inert today since both configs agree on encoding/gzip, but this fails startup loudly instead of letting a future divergence between the two configs silently corrupt responses on whichever mount lost the race. --- .env.example | 7 + core/factory.py | 3 +- core/pygeoapi-config-internal.yml | 293 ++++++++++++++++++++++++++++++ core/pygeoapi.py | 145 ++++++++++++--- 4 files changed, 425 insertions(+), 23 deletions(-) create mode 100644 core/pygeoapi-config-internal.yml diff --git a/.env.example b/.env.example index 3b53b9ae7..518028d1c 100644 --- a/.env.example +++ b/.env.example @@ -10,6 +10,13 @@ POSTGRES_PORT=5432 PYGEOAPI_POSTGRES_PASSWORD=your_password PYGEOAPI_POSTGRES_USER=your_username +# PYGEOAPI internal mount (/ogcapi-internal) -- authenticated, unfiltered +# (private/draft-inclusive) mirror of /ogcapi. Shares PYGEOAPI_POSTGRES_* +# above; only the mount path, runtime dir, and advertised server URL differ. +PYGEOAPI_INTERNAL_MOUNT_PATH=/ogcapi-internal +PYGEOAPI_INTERNAL_RUNTIME_DIR=/tmp/pygeoapi-internal +PYGEOAPI_INTERNAL_SERVER_URL= + # Connection pool configuration for parallel transfers # pool_size: number of persistent connections to maintain # max_overflow: additional connections allowed during peak usage diff --git a/core/factory.py b/core/factory.py index 69bcfba7e..029b0c2de 100644 --- a/core/factory.py +++ b/core/factory.py @@ -44,9 +44,10 @@ def create_api_app(): initialize_runtime() app = create_base_app() register_api_routes(app) - from core.pygeoapi import mount_pygeoapi + from core.pygeoapi import mount_pygeoapi, mount_pygeoapi_internal mount_pygeoapi(app) + mount_pygeoapi_internal(app) if os.environ.get("SESSION_SECRET_KEY"): configure_session_middleware(app) configure_cors_middleware(app) diff --git a/core/pygeoapi-config-internal.yml b/core/pygeoapi-config-internal.yml new file mode 100644 index 000000000..7bfbb1590 --- /dev/null +++ b/core/pygeoapi-config-internal.yml @@ -0,0 +1,293 @@ +server: + bind: + host: 0.0.0.0 + port: 8000 + url: {server_url} + mimetype: application/json; charset=UTF-8 + encoding: utf-8 + language: en-US + limits: + default_items: 10 + max_items: 10000 + map: + url: https://tile.openstreetmap.org/{{z}}/{{x}}/{{y}}.png + attribution: "© OpenStreetMap contributors" + +logging: + level: INFO + +metadata: + identification: + title: Ocotillo OGC API (Internal) + description: >- + Authenticated internal OGC API - Features backed by PostGIS and + pygeoapi. Unlike the public /ogcapi mount, these collections are not + filtered by release_status and include private and draft records. + keywords: [features, ogcapi, postgis, pygeoapi, internal] + terms_of_service: https://example.com/terms + url: https://example.com + license: + name: CC-BY 4.0 + url: https://creativecommons.org/licenses/by/4.0/ + provider: + name: NMBGMR + url: https://geoinfo.nmt.edu + contact: + name: API Support + email: support@example.com + +resources: + locations: + type: collection + title: Locations + description: Geographic locations and site coordinates used by Ocotillo features. + keywords: [locations] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_locations + geom_field: point + + latest_depth_to_water_wells: + type: collection + title: Latest Depth to Water (Water Wells) + description: Most recent depth-to-water below ground surface observation for each water well. + keywords: [water-wells, groundwater-level, depth-to-water-bgs, latest] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_latest_depth_to_water_wells + geom_field: point + + avg_tds_wells: + type: collection + title: Average TDS (Water Wells) + description: Average total dissolved solids (TDS) from major chemistry results for each water well. + keywords: [water-wells, chemistry, tds, total-dissolved-solids, average] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_avg_tds_wells + geom_field: point + + latest_tds_wells: + type: collection + title: Latest TDS (Water Wells) + description: Most recent total dissolved solids (TDS) result from major chemistry for each water well. + keywords: [water-wells, chemistry, tds, total-dissolved-solids, latest] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_latest_tds_wells + geom_field: point + + depth_to_water_trend_wells: + type: collection + title: Depth to Water Trend (Water Wells) + description: Trend classification for depth to water based on slope in feet per year. + keywords: [water-wells, groundwater-level, depth-to-water, trend, slope] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_depth_to_water_trend_wells + geom_field: point + + water_elevation_wells: + type: collection + title: Water Elevation (Water Wells) + description: Most recent water elevation per well calculated as elevation minus depth to water below ground surface. + keywords: [water-wells, groundwater-level, water-elevation, depth-to-water] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_water_elevation_wells + geom_field: point + + water_well_summary: + type: collection + title: Water Well Summary + description: Summary metrics per water well, including latest, min/max, and trend for water levels. + keywords: [water-wells, summary, groundwater-level, trend] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_water_well_summary + geom_field: point + + major_chemistry_results: + type: collection + title: Major Chemistry (Water Wells) + description: Latest major chemistry analyte values for water wells, represented as static analyte columns. + keywords: [water-wells, chemistry, analytes, major-chemistry] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_major_chemistry_results + geom_field: point + + minor_chemistry_wells: + type: collection + title: Minor Chemistry (Water Wells) + description: Latest minor/trace chemistry analyte values for water wells, represented as static analyte columns. + keywords: [water-wells, chemistry, analytes, minor-chemistry, trace-chemistry] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_minor_chemistry_wells + geom_field: point + + actively_monitored_wells: + type: collection + title: Actively Monitored Wells + description: Wells in the collaborative network currently flagged as actively monitored. + keywords: [water-wells, monitoring, collaborative-network, actively-monitored] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_actively_monitored_wells + geom_field: point + + project_areas: + type: collection + title: Project Areas + description: Project groups with polygon project-area boundaries. + keywords: [project-areas, groups, boundaries] + extents: + spatial: + bbox: [-109.05, 31.33, -103.00, 37.00] + crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 + providers: + - type: feature + name: PostgreSQL + data: + host: {postgres_host} + port: {postgres_port} + dbname: {postgres_db} + user: {postgres_user} + password: {postgres_password_env} + search_path: [public] + id_field: id + table: ogc_internal_project_areas + geom_field: project_area + +{thing_collections_block} diff --git a/core/pygeoapi.py b/core/pygeoapi.py index 7783af100..bc422d024 100644 --- a/core/pygeoapi.py +++ b/core/pygeoapi.py @@ -108,13 +108,17 @@ def _template_path() -> Path: return Path(__file__).resolve().parent / "pygeoapi-config.yml" -def _mount_path() -> str: - # Read and sanitize the configured mount path, defaulting to "/ogcapi". - path = (os.environ.get("PYGEOAPI_MOUNT_PATH", "/ogcapi") or "").strip() +def _internal_template_path() -> Path: + return Path(__file__).resolve().parent / "pygeoapi-config-internal.yml" + + +def _sanitized_mount_path(env_var: str, default: str) -> str: + # Read and sanitize the configured mount path, falling back to `default`. + path = (os.environ.get(env_var, default) or "").strip() # Treat empty or root ("/") values as invalid and fall back to the default. if path in {"", "/"}: - path = "/ogcapi" + path = default # Ensure a single leading slash. if not path.startswith("/"): @@ -127,21 +131,27 @@ def _mount_path() -> str: # Disallow traversal/current-directory segments. segments = [segment for segment in path.split("/") if segment] if any(segment in {".", ".."} for segment in segments): - raise ValueError( - "Invalid PYGEOAPI_MOUNT_PATH: traversal segments are not allowed." - ) + raise ValueError(f"Invalid {env_var}: traversal segments are not allowed.") # Allow only slash-delimited segments of alphanumerics, underscore, # or hyphen. if not re.fullmatch(r"/[A-Za-z0-9_-]+(?:/[A-Za-z0-9_-]+)*", path): raise ValueError( - "Invalid PYGEOAPI_MOUNT_PATH: only letters, numbers, underscores, " + f"Invalid {env_var}: only letters, numbers, underscores, " "hyphens, and slashes are allowed." ) return path +def _mount_path() -> str: + return _sanitized_mount_path("PYGEOAPI_MOUNT_PATH", "/ogcapi") + + +def _internal_mount_path() -> str: + return _sanitized_mount_path("PYGEOAPI_INTERNAL_MOUNT_PATH", "/ogcapi-internal") + + def _server_url() -> str: configured = os.environ.get("PYGEOAPI_SERVER_URL") if configured: @@ -149,10 +159,19 @@ def _server_url() -> str: return f"http://localhost:8000{_mount_path()}" -def _pygeoapi_dir() -> Path: +def _internal_server_url() -> str: + configured = os.environ.get("PYGEOAPI_INTERNAL_SERVER_URL") + if configured: + return configured.rstrip("/") + return f"http://localhost:8000{_internal_mount_path()}" + + +def _pygeoapi_dir( + runtime_dir_env: str = "PYGEOAPI_RUNTIME_DIR", default: str = "/tmp/pygeoapi" +) -> Path: # Use instance-local ephemeral storage by default (GAE-safe). - runtime_dir = (os.environ.get("PYGEOAPI_RUNTIME_DIR") or "").strip() - path = Path(runtime_dir) if runtime_dir else Path("/tmp/pygeoapi") + runtime_dir = (os.environ.get(runtime_dir_env) or "").strip() + path = Path(runtime_dir) if runtime_dir else Path(default) path.mkdir(parents=True, exist_ok=True) return path @@ -163,6 +182,7 @@ def _thing_collections_block( dbname: str, user: str, password_placeholder: str, + table_prefix: str = "ogc_", ) -> str: resources: dict[str, dict] = {} for collection in THING_COLLECTIONS: @@ -190,7 +210,7 @@ def _thing_collections_block( "search_path": ["public"], }, "id_field": "id", - "table": f"ogc_{collection['id']}", + "table": f"{table_prefix}{collection['id']}", "geom_field": "point", } ], @@ -237,11 +257,17 @@ def _pygeoapi_db_settings() -> tuple[str, str, str, str, str]: return host, port, dbname, user, "${PYGEOAPI_POSTGRES_PASSWORD}" -def _write_config(path: Path) -> None: +def _write_config( + path: Path, + *, + server_url: str, + table_prefix: str = "ogc_", + template_path: Path | None = None, +) -> None: host, port, dbname, user, password_placeholder = _pygeoapi_db_settings() - template = _template_path().read_text(encoding="utf-8") + template = (template_path or _template_path()).read_text(encoding="utf-8") config = template.format( - server_url=_server_url(), + server_url=server_url, postgres_host=host, postgres_port=port, postgres_db=dbname, @@ -253,14 +279,15 @@ def _write_config(path: Path) -> None: dbname=dbname, user=user, password_placeholder=password_placeholder, + table_prefix=table_prefix, ), ) - # NOTE: The generated runtime config file at - # `${PYGEOAPI_RUNTIME_DIR}/pygeoapi-config.yml` (default: - # `/tmp/pygeoapi/pygeoapi-config.yml`) contains database connection details - # (host, port, dbname, user). Although the password is expected to be - # provided via environment variables at runtime by pygeoapi, this file - # should still be treated as sensitive configuration: + # NOTE: The generated runtime config file (default: + # `/tmp/pygeoapi/pygeoapi-config.yml` or + # `/tmp/pygeoapi-internal/pygeoapi-config.yml`) contains database + # connection details (host, port, dbname, user). Although the password is + # expected to be provided via environment variables at runtime by + # pygeoapi, this file should still be treated as sensitive configuration: # * Do not commit it to version control. # * Do not expose it in logs, error messages, or diagnostics. # * Ensure filesystem permissions restrict access appropriately. @@ -268,6 +295,32 @@ def _write_config(path: Path) -> None: path.chmod(0o600) +def _assert_server_settings_match( + public_config_path: Path, internal_config_path: Path +) -> None: + # pygeoapi.api.API.__init__ mutates process-wide, module-level globals + # (CHARSET, FORMAT_TYPES) that persist across the importlib.reload this + # scheme relies on -- whichever mount is constructed last wins for both. + # Inert as long as both configs agree on these settings; fail loudly at + # startup rather than let a future divergence silently corrupt responses + # on whichever mount lost the race. + public_server = yaml.safe_load(public_config_path.read_text(encoding="utf-8")).get( + "server", {} + ) + internal_server = yaml.safe_load( + internal_config_path.read_text(encoding="utf-8") + ).get("server", {}) + for key in ("encoding", "gzip"): + if public_server.get(key) != internal_server.get(key): + raise RuntimeError( + "pygeoapi public/internal config drift detected: " + f"server.{key} differs ({public_server.get(key)!r} vs " + f"{internal_server.get(key)!r}). Both configs must agree " + "here since pygeoapi.api.API.__init__ mutates shared " + "process-wide globals from these settings." + ) + + def _generate_openapi(config_path: Path, openapi_path: Path) -> None: from pygeoapi.openapi import generate_openapi_document @@ -300,7 +353,7 @@ def mount_pygeoapi(app: FastAPI) -> None: pygeoapi_dir = _pygeoapi_dir() config_path = pygeoapi_dir / "pygeoapi-config.yml" openapi_path = pygeoapi_dir / "pygeoapi-openapi.yml" - _write_config(config_path) + _write_config(config_path, server_url=_server_url()) _generate_openapi(config_path, openapi_path) os.environ["PYGEOAPI_CONFIG"] = str(config_path) @@ -311,3 +364,51 @@ def mount_pygeoapi(app: FastAPI) -> None: app.mount(mount_path, pygeoapi_app) app.state.pygeoapi_mounted = True + + +def mount_pygeoapi_internal(app: FastAPI) -> None: + if getattr(app.state, "pygeoapi_internal_mounted", False): + return + if find_spec("pygeoapi") is None: + raise RuntimeError( + "pygeoapi is not installed. Rebuild/sync dependencies so " + "/ogcapi-internal can be mounted." + ) + + public_mount_path = _mount_path() + internal_mount_path = _internal_mount_path() + if internal_mount_path == public_mount_path: + # Starlette doesn't error on duplicate mount paths -- it registers + # both Mounts and matches whichever was registered first (the + # public mount), leaving the internal mount silently unreachable. + # Fail loudly at startup instead of that way blind. + raise RuntimeError( + "PYGEOAPI_MOUNT_PATH and PYGEOAPI_INTERNAL_MOUNT_PATH both " + f"resolve to {internal_mount_path!r}. They must be distinct." + ) + + internal_dir = _pygeoapi_dir( + "PYGEOAPI_INTERNAL_RUNTIME_DIR", "/tmp/pygeoapi-internal" + ) + config_path = internal_dir / "pygeoapi-config.yml" + openapi_path = internal_dir / "pygeoapi-openapi.yml" + _write_config( + config_path, + server_url=_internal_server_url(), + table_prefix="ogc_internal_", + template_path=_internal_template_path(), + ) + _generate_openapi(config_path, openapi_path) + _assert_server_settings_match(_pygeoapi_dir() / "pygeoapi-config.yml", config_path) + + os.environ["PYGEOAPI_CONFIG"] = str(config_path) + os.environ["PYGEOAPI_OPENAPI"] = str(openapi_path) + + pygeoapi_app = _load_pygeoapi_app() + + from core.internal_ogc_auth import InternalOGCAuthMiddleware + + app.add_middleware(InternalOGCAuthMiddleware, mount_path=internal_mount_path) + app.mount(internal_mount_path, pygeoapi_app) + + app.state.pygeoapi_internal_mounted = True From 9e589ef993ff8005b65d4c4a20506970b02805e1 Mon Sep 17 00:00:00 2001 From: Kelsey Smuczynski Date: Fri, 17 Jul 2026 23:20:33 -0600 Subject: [PATCH 3/7] feat(ogc): add unfiltered internal OGC views Mirrors all 22 ogc_* relations f4a5b6c7d8e9 filters, as ogc_internal_* counterparts with no release_status predicate, so /ogcapi-internal can serve private/draft records to authenticated staff. Reuses f4a5b6c7d8e9's parametrized builder functions (public_only kept in the signature and always passed False) for structural parity rather than a differently-shaped rewrite, since that's what the drift-detection test in the next commit diffs against. This repo keeps migrations self-contained with no cross-migration imports, so the two chemistry views' analyte-mapping CASE blocks are duplicated here rather than shared via a helper module -- a shared module would mean a future edit to one ticket's migration silently changes what the other replays from scratch. Added a one-line cross-reference comment in both files pointing at each other and the parity test. ogc_internal_locations has no release_status predicate at all (unlike ogc_locations, which is always public-only even on its own downgrade path) since it never existed in any filtered form before this migration. downgrade() simply drops all 22 relations rather than restoring a prior state, since none of them existed before this migration. --- .../2d3c3a268652_create_internal_ogc_views.py | 1304 +++++++++++++++++ ...blic_release_status_filter_to_ogc_views.py | 6 + 2 files changed, 1310 insertions(+) create mode 100644 alembic/versions/2d3c3a268652_create_internal_ogc_views.py diff --git a/alembic/versions/2d3c3a268652_create_internal_ogc_views.py b/alembic/versions/2d3c3a268652_create_internal_ogc_views.py new file mode 100644 index 000000000..affda020d --- /dev/null +++ b/alembic/versions/2d3c3a268652_create_internal_ogc_views.py @@ -0,0 +1,1304 @@ +"""create internal ogc views + +Companion migration to f4a5b6c7d8e9 (public release_status filter on ogc_* +views): creates a second, unfiltered copy of the same 22 relations, named +ogc_internal_, backing the authenticated /ogcapi-internal mount +(core/pygeoapi.py::mount_pygeoapi_internal). Full parity with the public +set, per ticket A11 -- not a subset. + +The major/minor chemistry analyte-mapping CASE blocks and +STATIC_ANALYTE_COLUMNS lists below are intentionally character-for-character +identical (modulo view name) to their counterparts in f4a5b6c7d8e9 -- this +codebase keeps migrations self-contained with no cross-migration imports, so +the logic is duplicated here rather than shared. tests/test_migration_view_ +parity.py enforces the two stay in sync: if you fix an analyte mapping in +one file, apply the same fix to the other. + +ogc_internal_locations has no release_status predicate at all (unlike +ogc_locations, which is always public-only) -- the internal mount is +unfiltered by design, and ogc_internal_locations never existed before this +migration in any form. + +ogc_internal_actively_monitored_wells gets no predicate of its own -- like +its public counterpart, it inherits whichever rows ogc_internal_water_well_ +summary exposes (here, all of them) transitively via a direct JOIN. Because +of that JOIN, it must be dropped before ogc_internal_water_well_summary and +recreated after (same ordering constraint as the public side). + +All 22 relations here are newly created by this migration -- none of them +existed in any form beforehand -- so downgrade() simply drops them rather +than recreating a prior state. + +Revision ID: 2d3c3a268652 +Revises: f4a5b6c7d8e9 +Create Date: 2026-07-16 00:00:00.000000 +""" + +import re +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import inspect, text + +# revision identifiers, used by Alembic. +revision: str = "2d3c3a268652" +down_revision: Union[str, Sequence[str], None] = "f4a5b6c7d8e9" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +REQUIRED_TABLES = { + "thing", + "location", + "location_thing_association", + "group", + "group_thing_association", + "status_history", + "observation", + "sample", + "field_activity", + "field_event", + "data_provenance", + "NMA_MajorChemistry", + "NMA_Chemistry_SampleInfo", + "NMA_MinorTraceChemistry", +} + +LATEST_LOCATION_CTE = """ +SELECT DISTINCT ON (lta.thing_id) + lta.thing_id, + lta.location_id, + lta.effective_start +FROM location_thing_association AS lta +WHERE lta.effective_end IS NULL +ORDER BY lta.thing_id, lta.effective_start DESC +""".strip() + +# Same 11 thing-type views as f4a5b6c7d8e9's THING_VIEWS. +THING_VIEWS = [ + ("water_wells", "water well"), + ("springs", "spring"), + ("diversions_surface_water", "diversion of surface water, etc."), + ("ephemeral_streams", "ephemeral stream"), + ("lakes_ponds_reservoirs", "lake, pond or reservoir"), + ("meteorological_stations", "meteorological station"), + ("other_things", "other"), + ("outfalls_wastewater_return_flow", "outfall of wastewater or return flow"), + ("perennial_streams", "perennial stream"), + ("rock_sample_locations", "rock sample location"), + ("soil_gas_sample_locations", "soil gas sample location"), +] + + +def _safe_view_id(view_id: str) -> str: + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", view_id): + raise ValueError(f"Unsafe view id: {view_id!r}") + return view_id + + +def _drop_view_or_materialized_view(view_name: str) -> None: + # DROP VIEW IF EXISTS / DROP MATERIALIZED VIEW IF EXISTS only suppress + # "relation does not exist" -- Postgres still raises WrongObjectType if + # the relation exists as the other kind (e.g. DROP VIEW against an + # existing materialized view), so the relation's actual kind must be + # checked first rather than trying both blindly. + bind = op.get_bind() + relkind = bind.execute( + text("SELECT relkind FROM pg_class WHERE oid = to_regclass(:name)"), + {"name": view_name}, + ).scalar() + if relkind == "m": + op.execute(text(f"DROP MATERIALIZED VIEW IF EXISTS {view_name}")) + elif relkind == "v": + op.execute(text(f"DROP VIEW IF EXISTS {view_name}")) + + +def _check_required_tables() -> None: + bind = op.get_bind() + inspector = inspect(bind) + existing_tables = set(inspector.get_table_names(schema="public")) + missing = REQUIRED_TABLES - existing_tables + if missing: + raise RuntimeError( + "Cannot create internal OGC views. " + f"Missing required tables: {', '.join(sorted(missing))}" + ) + + +def _create_thing_view(view_id: str, thing_type: str, public_only: bool) -> str: + safe_view_id = _safe_view_id(view_id) + escaped_thing_type = thing_type.replace("'", "''") + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE VIEW ogc_internal_{safe_view_id} AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ) + SELECT + t.id, + t.name, + t.first_visit_date, + t.nma_pk_welldata, + t.well_depth, + t.hole_depth, + t.well_casing_diameter, + t.well_casing_depth, + t.well_completion_date, + t.well_driller_name, + t.well_construction_method, + t.well_pump_type, + t.well_pump_depth, + t.formation_completion_code, + t.nma_formation_zone, + t.release_status, + l.elevation, + l.point + FROM thing AS t + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + WHERE t.thing_type = '{escaped_thing_type}'{release_filter} + """ + + +def _create_latest_depth_view(public_only: bool) -> str: + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE MATERIALIZED VIEW ogc_internal_latest_depth_to_water_wells AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + ranked_obs AS ( + SELECT + fe.thing_id, + o.id AS observation_id, + o.observation_datetime, + o.value, + o.measuring_point_height, + -- Treat NULL measuring_point_height as 0 when computing + -- depth_to_water_bgs. + ( + o.value - COALESCE(o.measuring_point_height, 0) + ) AS depth_to_water_bgs, + ROW_NUMBER() OVER ( + PARTITION BY fe.thing_id + ORDER BY o.observation_datetime DESC, o.id DESC + ) AS rn + FROM observation AS o + JOIN sample AS s ON s.id = o.sample_id + JOIN field_activity AS fa ON fa.id = s.field_activity_id + JOIN field_event AS fe ON fe.id = fa.field_event_id + JOIN thing AS t ON t.id = fe.thing_id + WHERE + t.thing_type = 'water well' + AND fa.activity_type = 'groundwater level' + AND o.value IS NOT NULL{release_filter} + ) + SELECT + t.id AS id, + t.name, + t.thing_type, + ro.observation_id, + ro.observation_datetime, + ro.value AS depth_to_water_reference, + ro.measuring_point_height, + ro.depth_to_water_bgs, + l.point + FROM ranked_obs AS ro + JOIN thing AS t ON t.id = ro.thing_id + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + WHERE ro.rn = 1 + """ + + +def _create_avg_tds_view(public_only: bool) -> str: + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE MATERIALIZED VIEW ogc_internal_avg_tds_wells AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + tds_obs AS ( + SELECT + csi.thing_id, + mc.id AS major_chemistry_id, + COALESCE(mc."AnalysisDate", csi."CollectionDate")::date AS observation_date, + mc."SampleValue" AS sample_value, + mc."Units" AS units + FROM "NMA_MajorChemistry" AS mc + JOIN "NMA_Chemistry_SampleInfo" AS csi + ON csi.id = mc.chemistry_sample_info_id + JOIN thing AS t ON t.id = csi.thing_id + WHERE + t.thing_type = 'water well' + AND mc."SampleValue" IS NOT NULL + AND ( + lower(coalesce(mc."Analyte", '')) IN ( + 'tds', + 'total dissolved solids' + ) + OR lower(coalesce(mc."Symbol", '')) = 'tds' + ){release_filter} + ) + SELECT + t.id AS id, + t.name, + t.thing_type, + COUNT(to2.major_chemistry_id)::integer AS tds_observation_count, + AVG(to2.sample_value)::double precision AS avg_tds_value, + MIN(to2.observation_date) AS first_tds_observation_date, + MAX(to2.observation_date) AS last_tds_observation_date, + l.point + FROM tds_obs AS to2 + JOIN thing AS t ON t.id = to2.thing_id + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + GROUP BY t.id, t.name, t.thing_type, l.point + """ + + +def _create_latest_tds_view(public_only: bool) -> str: + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE VIEW ogc_internal_latest_tds_wells AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + tds_obs AS ( + SELECT + csi.thing_id, + mc.id AS major_chemistry_id, + COALESCE(mc."AnalysisDate", csi."CollectionDate") AS observation_datetime, + mc."SampleValue" AS sample_value, + mc."Units" AS units + FROM "NMA_MajorChemistry" AS mc + JOIN "NMA_Chemistry_SampleInfo" AS csi + ON csi.id = mc.chemistry_sample_info_id + JOIN thing AS t ON t.id = csi.thing_id + WHERE + t.thing_type = 'water well' + AND mc."SampleValue" IS NOT NULL + AND ( + lower(coalesce(mc."Analyte", '')) IN ( + 'tds', + 'total dissolved solids' + ) + OR lower(coalesce(mc."Symbol", '')) = 'tds' + ){release_filter} + ), + ranked_tds AS ( + SELECT + to2.thing_id, + to2.major_chemistry_id, + to2.observation_datetime, + to2.sample_value, + to2.units, + ROW_NUMBER() OVER ( + PARTITION BY to2.thing_id + ORDER BY to2.observation_datetime DESC NULLS LAST, to2.major_chemistry_id DESC + ) AS rn + FROM tds_obs AS to2 + ) + SELECT + t.id AS id, + t.name, + t.thing_type, + rt.major_chemistry_id, + rt.observation_datetime::date AS latest_tds_observation_date, + rt.sample_value AS latest_tds_value, + rt.units AS latest_tds_units, + l.point + FROM ranked_tds AS rt + JOIN thing AS t ON t.id = rt.thing_id + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + WHERE rt.rn = 1 + """ + + +def _create_depth_to_water_trend_view(public_only: bool) -> str: + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE MATERIALIZED VIEW ogc_internal_depth_to_water_trend_wells AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + obs AS ( + SELECT + fe.thing_id, + o.observation_datetime, + (o.value - COALESCE(o.measuring_point_height, 0)) AS depth_to_water_bgs + FROM observation AS o + JOIN sample AS s ON s.id = o.sample_id + JOIN field_activity AS fa ON fa.id = s.field_activity_id + JOIN field_event AS fe ON fe.id = fa.field_event_id + JOIN thing AS t ON t.id = fe.thing_id + WHERE + t.thing_type = 'water well' + AND fa.activity_type = 'groundwater level' + AND o.value IS NOT NULL + AND o.observation_datetime IS NOT NULL{release_filter} + ), + agg AS ( + SELECT + ob.thing_id, + COUNT(*)::integer AS record_count, + MIN(ob.observation_datetime) AS first_observation_datetime, + MAX(ob.observation_datetime) AS last_observation_datetime, + EXTRACT(EPOCH FROM (MAX(ob.observation_datetime) - MIN(ob.observation_datetime))) + / 31557600.0 AS span_years, + REGR_SLOPE( + ob.depth_to_water_bgs, + EXTRACT(EPOCH FROM ob.observation_datetime) + ) * 31557600.0 AS slope_ft_per_year + FROM obs AS ob + GROUP BY ob.thing_id + ) + SELECT + t.id AS id, + t.name, + t.thing_type, + a.record_count, + a.first_observation_datetime, + a.last_observation_datetime, + a.span_years, + a.slope_ft_per_year, + CASE + WHEN a.record_count >= 10 OR (a.record_count >= 4 AND a.span_years >= 2.0) THEN + CASE + WHEN a.slope_ft_per_year IS NULL THEN 'not enough data' + WHEN a.slope_ft_per_year > 0.25 THEN 'increasing' + WHEN a.slope_ft_per_year < -0.25 THEN 'decreasing' + ELSE 'stable' + END + ELSE 'not enough data' + END AS trend_category, + l.point + FROM agg AS a + JOIN thing AS t ON t.id = a.thing_id + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + """ + + +def _create_water_well_summary_view(public_only: bool) -> str: + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE MATERIALIZED VIEW ogc_internal_water_well_summary AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + wl_obs AS ( + SELECT + fe.thing_id, + o.id AS observation_id, + o.observation_datetime, + (o.value - COALESCE(o.measuring_point_height, 0)) AS water_level + FROM observation AS o + JOIN sample AS s ON s.id = o.sample_id + JOIN field_activity AS fa ON fa.id = s.field_activity_id + JOIN field_event AS fe ON fe.id = fa.field_event_id + JOIN thing AS t ON t.id = fe.thing_id + WHERE + t.thing_type = 'water well' + AND fa.activity_type = 'groundwater level' + AND o.value IS NOT NULL + AND o.observation_datetime IS NOT NULL{release_filter} + ), + wl_agg AS ( + SELECT + w.thing_id, + COUNT(*)::integer AS total_water_levels, + MIN(w.water_level) AS min_water_level, + MAX(w.water_level) AS max_water_level, + REGR_SLOPE( + w.water_level, + EXTRACT(EPOCH FROM w.observation_datetime) + ) * 31557600.0 AS water_level_trend_ft_per_year + FROM wl_obs AS w + GROUP BY w.thing_id + ), + wl_last AS ( + SELECT + ranked.thing_id, + ranked.water_level AS last_water_level, + ranked.observation_datetime AS last_water_level_datetime + FROM ( + SELECT + w.thing_id, + w.water_level, + w.observation_datetime, + ROW_NUMBER() OVER ( + PARTITION BY w.thing_id + ORDER BY w.observation_datetime DESC, w.observation_id DESC + ) AS rn + FROM wl_obs AS w + ) AS ranked + WHERE ranked.rn = 1 + ) + SELECT + t.id AS id, + t.name, + t.well_depth, + l.elevation, + dpl.collection_method AS elevation_method, + t.nma_formation_zone AS formation_zone, + wa.total_water_levels, + wl.last_water_level, + wl.last_water_level_datetime, + wa.min_water_level, + wa.max_water_level, + wa.water_level_trend_ft_per_year, + l.point + FROM thing AS t + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + JOIN wl_agg AS wa ON wa.thing_id = t.id + LEFT JOIN wl_last AS wl ON wl.thing_id = t.id + LEFT JOIN LATERAL ( + SELECT dp.collection_method + FROM data_provenance AS dp + WHERE + dp.target_table = 'location' + AND dp.target_id = l.id + AND dp.field_name = 'elevation' + ORDER BY dp.id DESC + LIMIT 1 + ) AS dpl ON true + WHERE t.thing_type = 'water well' + AND wa.total_water_levels > 0 + """ + + +# Static analyte columns for major chemistry pivots. +# Includes aliases observed in current DB values (e.g., Ca(total), IONBAL, TAn, TCat, Na+K). +# Kept character-for-character identical to f4a5b6c7d8e9's copy -- see +# tests/test_migration_view_parity.py. +STATIC_ANALYTE_COLUMNS_MAJOR: list[tuple[str, str]] = [ + ("tds", "tds"), + ("calcium", "calcium"), + ("calcium_total", "calcium_total"), + ("magnesium", "magnesium"), + ("magnesium_total", "magnesium_total"), + ("sodium", "sodium"), + ("sodium_total", "sodium_total"), + ("potassium", "potassium"), + ("potassium_total", "potassium_total"), + ("sodium_plus_potassium", "sodium_plus_potassium"), + ("bicarbonate", "bicarbonate"), + ("carbonate", "carbonate"), + ("sulfate", "sulfate"), + ("chloride", "chloride"), + ("ion_balance", "ion_balance"), + ("total_anions", "total_anions"), + ("total_cations", "total_cations"), + ("alkalinity", "alkalinity"), + ("hardness", "hardness"), + ("specific_conductance", "specific_conductance"), + ("ph", "ph"), + ("nitrate", "nitrate"), + ("fluoride", "fluoride"), + ("silica", "silica"), +] + + +def _major_chemistry_select_columns() -> str: + return ",\n".join( + [ + ( + " MAX(lr.sample_value) FILTER " + f"(WHERE lr.analyte_key = '{analyte_key}') AS {column_name}" + ) + for analyte_key, column_name in STATIC_ANALYTE_COLUMNS_MAJOR + ] + ) + + +def _major_chemistry_unit_columns() -> str: + return ",\n".join( + [ + ( + " MAX(lr.units) FILTER " + f"(WHERE lr.analyte_key = '{analyte_key}') AS {column_name}_units" + ) + for analyte_key, column_name in STATIC_ANALYTE_COLUMNS_MAJOR + ] + ) + + +def _create_major_chemistry_results_view(public_only: bool) -> str: + static_columns = _major_chemistry_select_columns() + static_unit_columns = _major_chemistry_unit_columns() + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE MATERIALIZED VIEW ogc_internal_major_chemistry_results AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + chemistry_rows AS ( + SELECT + csi.thing_id, + mc.id AS result_id, + COALESCE(mc."AnalysisDate", csi."CollectionDate") AS observation_datetime, + trim(mc."Analyte") AS analyte_name, + trim(mc."Symbol") AS symbol_name, + mc."SampleValue"::double precision AS sample_value, + mc."Units" AS units + FROM "NMA_MajorChemistry" AS mc + JOIN "NMA_Chemistry_SampleInfo" AS csi + ON csi.id = mc.chemistry_sample_info_id + JOIN thing AS t + ON t.id = csi.thing_id + WHERE mc."SampleValue" IS NOT NULL + AND t.thing_type = 'water well'{release_filter} + ), + normalized_rows AS ( + SELECT + cr.thing_id, + cr.result_id, + cr.observation_datetime, + NULLIF( + regexp_replace( + lower(trim(coalesce(cr.analyte_name, ''))), + '[^a-z0-9]+', + '', + 'g' + ), + '' + ) AS analyte_token, + NULLIF( + regexp_replace( + lower(trim(coalesce(cr.symbol_name, ''))), + '[^a-z0-9]+', + '', + 'g' + ), + '' + ) AS symbol_token, + cr.sample_value, + cr.units + FROM chemistry_rows AS cr + ), + mapped_rows AS ( + SELECT + nr.thing_id, + nr.result_id, + nr.observation_datetime, + CASE + WHEN coalesce(nr.symbol_token, '') = 'tds' + OR coalesce(nr.analyte_token, '') IN ('tds', 'totaldissolvedsolids') + THEN 'tds' + + WHEN coalesce(nr.symbol_token, '') = 'ca' + OR coalesce(nr.analyte_token, '') = 'ca' + THEN 'calcium' + WHEN coalesce(nr.analyte_token, '') = 'catotal' + THEN 'calcium_total' + + WHEN coalesce(nr.symbol_token, '') = 'mg' + OR coalesce(nr.analyte_token, '') = 'mg' + THEN 'magnesium' + WHEN coalesce(nr.analyte_token, '') = 'mgtotal' + THEN 'magnesium_total' + + WHEN coalesce(nr.symbol_token, '') = 'na' + OR coalesce(nr.analyte_token, '') = 'na' + THEN 'sodium' + WHEN coalesce(nr.analyte_token, '') = 'natotal' + THEN 'sodium_total' + + WHEN coalesce(nr.symbol_token, '') = 'k' + OR coalesce(nr.analyte_token, '') = 'k' + THEN 'potassium' + WHEN coalesce(nr.analyte_token, '') = 'ktotal' + THEN 'potassium_total' + + WHEN coalesce(nr.analyte_token, '') = 'nak' + THEN 'sodium_plus_potassium' + + WHEN coalesce(nr.symbol_token, '') = 'hco3' + OR coalesce(nr.analyte_token, '') = 'hco3' + THEN 'bicarbonate' + WHEN coalesce(nr.symbol_token, '') = 'co3' + OR coalesce(nr.analyte_token, '') = 'co3' + THEN 'carbonate' + WHEN coalesce(nr.symbol_token, '') = 'so4' + OR coalesce(nr.analyte_token, '') = 'so4' + THEN 'sulfate' + WHEN coalesce(nr.symbol_token, '') = 'cl' + OR coalesce(nr.analyte_token, '') = 'cl' + THEN 'chloride' + + WHEN coalesce(nr.analyte_token, '') = 'ionbal' + THEN 'ion_balance' + WHEN coalesce(nr.analyte_token, '') = 'tan' + THEN 'total_anions' + WHEN coalesce(nr.analyte_token, '') = 'tcat' + THEN 'total_cations' + + WHEN coalesce(nr.analyte_token, '') IN ('alk', 'alkalinity') + THEN 'alkalinity' + WHEN coalesce(nr.analyte_token, '') IN ('hrd', 'hardness') + THEN 'hardness' + WHEN coalesce(nr.analyte_token, '') IN ( + 'condlab', + 'specificconductance', + 'specificconductivity', + 'conductivity' + ) + THEN 'specific_conductance' + WHEN coalesce(nr.symbol_token, '') = 'ph' + OR coalesce(nr.analyte_token, '') IN ('ph', 'phl') + THEN 'ph' + + WHEN coalesce(nr.symbol_token, '') = 'no3' + OR coalesce(nr.analyte_token, '') IN ('no3', 'nitrate') + THEN 'nitrate' + WHEN coalesce(nr.symbol_token, '') = 'f' + OR coalesce(nr.analyte_token, '') IN ('f', 'fluoride') + THEN 'fluoride' + WHEN coalesce(nr.symbol_token, '') = 'sio2' + OR coalesce(nr.analyte_token, '') IN ('sio2', 'silica') + THEN 'silica' + + ELSE NULL + END AS analyte_key, + nr.sample_value, + nr.units + FROM normalized_rows AS nr + ), + latest_results AS ( + SELECT + mr.thing_id, + mr.analyte_key, + mr.sample_value, + mr.units, + mr.observation_datetime, + ROW_NUMBER() OVER ( + PARTITION BY mr.thing_id, mr.analyte_key + ORDER BY mr.observation_datetime DESC NULLS LAST, mr.result_id DESC + ) AS rn + FROM mapped_rows AS mr + WHERE mr.analyte_key IS NOT NULL + ) + SELECT + t.id AS id, + ll.location_id, + t.name, + t.thing_type, + COUNT(*)::integer AS analyte_count, + MAX(lr.observation_datetime::date) AS latest_chemistry_date, +{static_columns}, +{static_unit_columns}, + l.point + FROM latest_results AS lr + JOIN thing AS t ON t.id = lr.thing_id + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + WHERE lr.rn = 1 + GROUP BY t.id, ll.location_id, t.name, t.thing_type, l.point + """ + + +# Kept character-for-character identical to f4a5b6c7d8e9's copy -- see +# tests/test_migration_view_parity.py. +STATIC_ANALYTE_COLUMNS_MINOR: list[tuple[str, str]] = [ + ("h2r", "h2r"), + ("o18r", "o18r"), + ("c13r", "c13r"), + ("c14", "c14"), + ("c14_years", "c14_years"), + ("fluoride", "fluoride"), + ("barium", "barium"), + ("barium_total", "barium_total"), + ("copper", "copper"), + ("copper_total", "copper_total"), + ("zinc", "zinc"), + ("zinc_total", "zinc_total"), + ("molybdenum", "molybdenum"), + ("molybdenum_total", "molybdenum_total"), + ("silica", "silica"), + ("silicon", "silicon"), + ("silicon_total", "silicon_total"), + ("manganese", "manganese"), + ("manganese_total", "manganese_total"), + ("iron", "iron"), + ("iron_total", "iron_total"), + ("strontium", "strontium"), + ("strontium_total", "strontium_total"), + ("chromium", "chromium"), + ("chromium_total", "chromium_total"), + ("boron", "boron"), + ("boron_total", "boron_total"), + ("uranium", "uranium"), + ("uranium_total", "uranium_total"), + ("lithium", "lithium"), + ("lithium_total", "lithium_total"), + ("silver", "silver"), + ("silver_total", "silver_total"), + ("antimony", "antimony"), + ("antimony_total", "antimony_total"), + ("beryllium", "beryllium"), + ("beryllium_total", "beryllium_total"), + ("lead", "lead"), + ("lead_total", "lead_total"), + ("thallium", "thallium"), + ("thallium_total", "thallium_total"), + ("bromide", "bromide"), + ("selenium", "selenium"), + ("selenium_total", "selenium_total"), + ("vanadium", "vanadium"), + ("vanadium_total", "vanadium_total"), + ("aluminum", "aluminum"), + ("aluminum_total", "aluminum_total"), + ("arsenic", "arsenic"), + ("arsenic_total", "arsenic_total"), + ("nickel", "nickel"), + ("nickel_total", "nickel_total"), + ("cadmium", "cadmium"), + ("cadmium_total", "cadmium_total"), + ("cobalt", "cobalt"), + ("cobalt_total", "cobalt_total"), + ("phosphate", "phosphate"), + ("nitrite", "nitrite"), + ("nitrate", "nitrate"), + ("nitrate_as_n", "nitrate_as_n"), + ("thorium", "thorium"), + ("thorium_total", "thorium_total"), + ("tin", "tin"), + ("tin_total", "tin_total"), + ("mercury", "mercury"), + ("mercury_total", "mercury_total"), + ("titanium", "titanium"), + ("titanium_total", "titanium_total"), +] + + +def _minor_chemistry_value_columns() -> str: + return ",\n".join( + [ + ( + " MAX(lr.sample_value) FILTER " + f"(WHERE lr.analyte_key = '{analyte_key}') AS {column_name}" + ) + for analyte_key, column_name in STATIC_ANALYTE_COLUMNS_MINOR + ] + ) + + +def _minor_chemistry_unit_columns() -> str: + return ",\n".join( + [ + ( + " MAX(lr.units) FILTER " + f"(WHERE lr.analyte_key = '{analyte_key}') AS {column_name}_units" + ) + for analyte_key, column_name in STATIC_ANALYTE_COLUMNS_MINOR + ] + ) + + +def _create_minor_chemistry_wells_view(public_only: bool) -> str: + value_columns = _minor_chemistry_value_columns() + unit_columns = _minor_chemistry_unit_columns() + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE MATERIALIZED VIEW ogc_internal_minor_chemistry_wells AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + chemistry_rows AS ( + SELECT + csi.thing_id, + mtc.id AS result_id, + COALESCE(mtc.analysis_date::timestamp, csi."CollectionDate") AS observation_datetime, + trim(mtc.analyte) AS analyte_name, + mtc.sample_value::double precision AS sample_value, + mtc.units AS units + FROM "NMA_MinorTraceChemistry" AS mtc + JOIN "NMA_Chemistry_SampleInfo" AS csi + ON csi.id = mtc.chemistry_sample_info_id + JOIN thing AS t ON t.id = csi.thing_id + WHERE + mtc.sample_value IS NOT NULL + AND t.thing_type = 'water well'{release_filter} + ), + normalized_rows AS ( + SELECT + cr.thing_id, + cr.result_id, + cr.observation_datetime, + NULLIF( + regexp_replace( + lower(trim(coalesce(cr.analyte_name, ''))), + '[^a-z0-9]+', + '', + 'g' + ), + '' + ) AS analyte_token, + cr.sample_value, + cr.units + FROM chemistry_rows AS cr + ), + mapped_rows AS ( + SELECT + nr.thing_id, + nr.result_id, + nr.observation_datetime, + CASE + WHEN coalesce(nr.analyte_token, '') = 'h2r' THEN 'h2r' + WHEN coalesce(nr.analyte_token, '') = 'o18r' THEN 'o18r' + WHEN coalesce(nr.analyte_token, '') = 'c13r' THEN 'c13r' + WHEN coalesce(nr.analyte_token, '') = 'c14' THEN 'c14' + WHEN coalesce(nr.analyte_token, '') = 'c14years' THEN 'c14_years' + + WHEN coalesce(nr.analyte_token, '') = 'f' THEN 'fluoride' + WHEN coalesce(nr.analyte_token, '') = 'ba' THEN 'barium' + WHEN coalesce(nr.analyte_token, '') = 'batotal' THEN 'barium_total' + WHEN coalesce(nr.analyte_token, '') = 'cu' THEN 'copper' + WHEN coalesce(nr.analyte_token, '') = 'cutotal' THEN 'copper_total' + WHEN coalesce(nr.analyte_token, '') = 'zn' THEN 'zinc' + WHEN coalesce(nr.analyte_token, '') = 'zntotal' THEN 'zinc_total' + WHEN coalesce(nr.analyte_token, '') = 'mo' THEN 'molybdenum' + WHEN coalesce(nr.analyte_token, '') = 'mototal' THEN 'molybdenum_total' + WHEN coalesce(nr.analyte_token, '') = 'sio2' THEN 'silica' + WHEN coalesce(nr.analyte_token, '') = 'si' THEN 'silicon' + WHEN coalesce(nr.analyte_token, '') = 'sitotal' THEN 'silicon_total' + WHEN coalesce(nr.analyte_token, '') = 'mn' THEN 'manganese' + WHEN coalesce(nr.analyte_token, '') = 'mntotal' THEN 'manganese_total' + WHEN coalesce(nr.analyte_token, '') = 'fe' THEN 'iron' + WHEN coalesce(nr.analyte_token, '') = 'fetotal' THEN 'iron_total' + WHEN coalesce(nr.analyte_token, '') = 'sr' THEN 'strontium' + WHEN coalesce(nr.analyte_token, '') = 'srtotal' THEN 'strontium_total' + WHEN coalesce(nr.analyte_token, '') = 'cr' THEN 'chromium' + WHEN coalesce(nr.analyte_token, '') = 'crtotal' THEN 'chromium_total' + WHEN coalesce(nr.analyte_token, '') = 'b' THEN 'boron' + WHEN coalesce(nr.analyte_token, '') = 'btotal' THEN 'boron_total' + WHEN coalesce(nr.analyte_token, '') = 'u' THEN 'uranium' + WHEN coalesce(nr.analyte_token, '') = 'utotal' THEN 'uranium_total' + WHEN coalesce(nr.analyte_token, '') = 'li' THEN 'lithium' + WHEN coalesce(nr.analyte_token, '') = 'litotal' THEN 'lithium_total' + WHEN coalesce(nr.analyte_token, '') = 'ag' THEN 'silver' + WHEN coalesce(nr.analyte_token, '') = 'agtotal' THEN 'silver_total' + WHEN coalesce(nr.analyte_token, '') = 'sb' THEN 'antimony' + WHEN coalesce(nr.analyte_token, '') = 'sbtotal' THEN 'antimony_total' + WHEN coalesce(nr.analyte_token, '') = 'be' THEN 'beryllium' + WHEN coalesce(nr.analyte_token, '') = 'betotal' THEN 'beryllium_total' + WHEN coalesce(nr.analyte_token, '') = 'pb' THEN 'lead' + WHEN coalesce(nr.analyte_token, '') = 'pbtotal' THEN 'lead_total' + WHEN coalesce(nr.analyte_token, '') = 'tl' THEN 'thallium' + WHEN coalesce(nr.analyte_token, '') = 'tltotal' THEN 'thallium_total' + WHEN coalesce(nr.analyte_token, '') = 'br' THEN 'bromide' + WHEN coalesce(nr.analyte_token, '') = 'se' THEN 'selenium' + WHEN coalesce(nr.analyte_token, '') = 'setotal' THEN 'selenium_total' + WHEN coalesce(nr.analyte_token, '') = 'v' THEN 'vanadium' + WHEN coalesce(nr.analyte_token, '') = 'vtotal' THEN 'vanadium_total' + WHEN coalesce(nr.analyte_token, '') = 'al' THEN 'aluminum' + WHEN coalesce(nr.analyte_token, '') = 'altotal' THEN 'aluminum_total' + WHEN coalesce(nr.analyte_token, '') = 'as' THEN 'arsenic' + WHEN coalesce(nr.analyte_token, '') = 'astotal' THEN 'arsenic_total' + WHEN coalesce(nr.analyte_token, '') = 'ni' THEN 'nickel' + WHEN coalesce(nr.analyte_token, '') = 'nitotal' THEN 'nickel_total' + WHEN coalesce(nr.analyte_token, '') = 'cd' THEN 'cadmium' + WHEN coalesce(nr.analyte_token, '') = 'cdtotal' THEN 'cadmium_total' + WHEN coalesce(nr.analyte_token, '') = 'co' THEN 'cobalt' + WHEN coalesce(nr.analyte_token, '') = 'cototal' THEN 'cobalt_total' + WHEN coalesce(nr.analyte_token, '') = 'po4' THEN 'phosphate' + WHEN coalesce(nr.analyte_token, '') = 'no2' THEN 'nitrite' + WHEN coalesce(nr.analyte_token, '') = 'no3' THEN 'nitrate' + WHEN coalesce(nr.analyte_token, '') = 'no3n' THEN 'nitrate_as_n' + WHEN coalesce(nr.analyte_token, '') = 'th' THEN 'thorium' + WHEN coalesce(nr.analyte_token, '') = 'thtotal' THEN 'thorium_total' + WHEN coalesce(nr.analyte_token, '') = 'sn' THEN 'tin' + WHEN coalesce(nr.analyte_token, '') = 'sntotal' THEN 'tin_total' + WHEN coalesce(nr.analyte_token, '') = 'hg' THEN 'mercury' + WHEN coalesce(nr.analyte_token, '') = 'hgtotal' THEN 'mercury_total' + WHEN coalesce(nr.analyte_token, '') = 'ti' THEN 'titanium' + WHEN coalesce(nr.analyte_token, '') = 'titotal' THEN 'titanium_total' + ELSE NULL + END AS analyte_key, + nr.sample_value, + nr.units + FROM normalized_rows AS nr + ), + latest_results AS ( + SELECT + mr.thing_id, + mr.analyte_key, + mr.sample_value, + mr.units, + mr.observation_datetime, + ROW_NUMBER() OVER ( + PARTITION BY mr.thing_id, mr.analyte_key + ORDER BY mr.observation_datetime DESC NULLS LAST, mr.result_id DESC + ) AS rn + FROM mapped_rows AS mr + WHERE mr.analyte_key IS NOT NULL + ) + SELECT + t.id AS id, + ll.location_id, + t.name, + t.thing_type, + COUNT(*)::integer AS analyte_count, + MAX(lr.observation_datetime::date) AS latest_chemistry_date, +{value_columns}, +{unit_columns}, + l.point + FROM latest_results AS lr + JOIN thing AS t ON t.id = lr.thing_id + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + WHERE lr.rn = 1 + AND t.thing_type = 'water well' + GROUP BY t.id, ll.location_id, t.name, t.thing_type, l.point + """ + + +METERS_TO_FEET = 3.28084 + + +def _create_water_elevation_view(public_only: bool) -> str: + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE MATERIALIZED VIEW ogc_internal_water_elevation_wells AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ), + ranked_obs AS ( + SELECT + fe.thing_id, + o.id AS observation_id, + o.observation_datetime, + CASE + WHEN lower(trim(o.unit)) IN ('m', 'meter', 'meters', 'metre', 'metres') THEN + (o.value * {METERS_TO_FEET}) - COALESCE(o.measuring_point_height, 0) + WHEN lower(trim(o.unit)) IN ('ft', 'foot', 'feet') THEN + o.value - COALESCE(o.measuring_point_height, 0) + ELSE + NULL + END AS depth_to_water_below_ground_surface + FROM observation AS o + JOIN sample AS s ON s.id = o.sample_id + JOIN field_activity AS fa ON fa.id = s.field_activity_id + JOIN field_event AS fe ON fe.id = fa.field_event_id + JOIN thing AS t ON t.id = fe.thing_id + WHERE + t.thing_type = 'water well' + AND fa.activity_type = 'groundwater level' + AND o.value IS NOT NULL + AND o.observation_datetime IS NOT NULL + AND lower(trim(o.unit)) IN ( + 'm', + 'meter', + 'meters', + 'metre', + 'metres', + 'ft', + 'foot', + 'feet' + ){release_filter} + ), + latest_obs AS ( + SELECT + ro.*, + ROW_NUMBER() OVER ( + PARTITION BY ro.thing_id + ORDER BY ro.observation_datetime DESC, ro.observation_id DESC + ) AS rn + FROM ranked_obs AS ro + ) + SELECT + t.id AS id, + t.name, + t.thing_type, + lo.observation_id, + lo.observation_datetime, + l.elevation AS elevation_m, + lo.depth_to_water_below_ground_surface AS depth_to_water_below_ground_surface_ft, + ((l.elevation * {METERS_TO_FEET}) - lo.depth_to_water_below_ground_surface) + AS water_elevation_ft, + l.point + FROM latest_obs AS lo + JOIN thing AS t ON t.id = lo.thing_id + JOIN latest_location AS ll ON ll.thing_id = t.id + JOIN location AS l ON l.id = ll.location_id + WHERE lo.rn = 1 + """ + + +def _create_actively_monitored_wells_view() -> str: + # No predicate of its own -- inherits whatever rows + # ogc_internal_water_well_summary exposes (here, all of them) + # transitively via the JOIN below. Mirrors the public side's + # ogc_actively_monitored_wells, which likewise never filters on + # status_history.release_status directly (see + # w1x2y3z4a5b6_drop_child_release_filters_from_ngwmn_views.py for why). + return """ + CREATE VIEW ogc_internal_actively_monitored_wells AS + WITH latest_monitoring_status AS ( + SELECT DISTINCT ON (sh.target_id) + sh.target_id AS thing_id, + sh.status_value + FROM status_history AS sh + WHERE + sh.target_table = 'thing' + AND sh.status_type = 'Monitoring Status' + ORDER BY sh.target_id, sh.start_date DESC, sh.id DESC + ) + SELECT + wws.id, + wws.name, + 'water well'::text AS thing_type, + wws.well_depth, + wws.elevation, + wws.elevation_method, + wws.formation_zone, + wws.total_water_levels, + wws.last_water_level, + wws.last_water_level_datetime, + wws.min_water_level, + wws.max_water_level, + wws.water_level_trend_ft_per_year, + g.id AS group_id, + g.name AS group_name, + g.group_type, + wws.point + FROM "group" AS g + JOIN group_thing_association AS gta ON gta.group_id = g.id + JOIN ogc_internal_water_well_summary AS wws ON wws.id = gta.thing_id + JOIN latest_monitoring_status AS lms ON lms.thing_id = wws.id + WHERE lower(trim(g.name)) = 'water level network' + AND lms.status_value = 'Currently monitored' + """ + + +def _create_project_areas_view(public_only: bool) -> str: + release_filter = " AND g.release_status = 'public'" if public_only else "" + return f""" + CREATE VIEW ogc_internal_project_areas AS + SELECT + g.id, + g.name, + g.description, + g.group_type, + g.release_status, + g.project_area + FROM "group" AS g + WHERE g.project_area IS NOT NULL{release_filter} + """ + + +def _create_locations_view() -> str: + # Unlike ogc_locations (always public-only, even on the public side's + # downgrade path), ogc_internal_locations has no release_status + # predicate at all -- the internal mount is unfiltered by design, and + # this relation never existed in any form before this migration. + # Column list matches ogc_locations exactly; see + # f4a5b6c7d8e9_apply_public_release_status_filter_to_ogc_views.py's + # _create_locations_view() for the db/location.py column verification. + return """ + CREATE VIEW ogc_internal_locations AS + SELECT + l.id, + l.nma_pk_location, + l.description, + l.county, + l.state, + l.quad_name, + l.nma_location_notes, + l.nma_coordinate_notes, + l.nma_data_reliability, + l.nma_date_created, + l.nma_site_date, + l.release_status, + l.elevation, + l.point + FROM location AS l + """ + + +def _recreate_all_internal_views() -> None: + # ogc_internal_actively_monitored_wells depends on + # ogc_internal_water_well_summary via a direct JOIN; Postgres refuses to + # drop a materialized view while a dependent view exists, so it must go + # first and come back last -- same ordering constraint as the public side. + _drop_view_or_materialized_view("ogc_internal_actively_monitored_wells") + + for view_id, thing_type in THING_VIEWS: + _drop_view_or_materialized_view(f"ogc_internal_{_safe_view_id(view_id)}") + op.execute(text(_create_thing_view(view_id, thing_type, public_only=False))) + + _drop_view_or_materialized_view("ogc_internal_latest_depth_to_water_wells") + op.execute(text(_create_latest_depth_view(public_only=False))) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW ogc_internal_latest_depth_to_water_wells IS " + "'Unfiltered latest depth-to-water per well view for the internal pygeoapi mount.'" + ) + ) + op.execute( + text( + "CREATE UNIQUE INDEX ux_ogc_internal_latest_depth_to_water_wells_id " + "ON ogc_internal_latest_depth_to_water_wells (id)" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_avg_tds_wells") + op.execute(text(_create_avg_tds_view(public_only=False))) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW ogc_internal_avg_tds_wells IS " + "'Unfiltered average TDS per well from major chemistry results for the internal pygeoapi mount.'" + ) + ) + op.execute( + text( + "CREATE UNIQUE INDEX ux_ogc_internal_avg_tds_wells_id " + "ON ogc_internal_avg_tds_wells (id)" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_latest_tds_wells") + op.execute(text(_create_latest_tds_view(public_only=False))) + op.execute( + text( + "COMMENT ON VIEW ogc_internal_latest_tds_wells IS " + "'Unfiltered latest TDS per well from major chemistry results for the internal pygeoapi mount.'" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_depth_to_water_trend_wells") + op.execute(text(_create_depth_to_water_trend_view(public_only=False))) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW ogc_internal_depth_to_water_trend_wells IS " + "'Unfiltered depth-to-water trend classification for water wells, for the internal pygeoapi mount.'" + ) + ) + op.execute( + text( + "CREATE UNIQUE INDEX ux_ogc_internal_depth_to_water_trend_wells_id " + "ON ogc_internal_depth_to_water_trend_wells (id)" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_water_well_summary") + op.execute(text(_create_water_well_summary_view(public_only=False))) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW ogc_internal_water_well_summary IS " + "'Unfiltered summary statistics for water wells including water-level trend, for the internal pygeoapi mount.'" + ) + ) + op.execute( + text( + "CREATE UNIQUE INDEX ux_ogc_internal_water_well_summary_id " + "ON ogc_internal_water_well_summary (id)" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_major_chemistry_results") + op.execute(text(_create_major_chemistry_results_view(public_only=False))) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW ogc_internal_major_chemistry_results IS " + "'Unfiltered latest major-chemistry analyte values per location, pivoted into static analyte columns, for the internal pygeoapi mount.'" + ) + ) + op.execute( + text( + "CREATE UNIQUE INDEX ux_ogc_internal_major_chemistry_results_id " + "ON ogc_internal_major_chemistry_results (id)" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_minor_chemistry_wells") + op.execute(text(_create_minor_chemistry_wells_view(public_only=False))) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW ogc_internal_minor_chemistry_wells IS " + "'Unfiltered latest minor/trace chemistry analyte values for water wells, pivoted into static analyte columns, for the internal pygeoapi mount.'" + ) + ) + op.execute( + text( + "CREATE UNIQUE INDEX ux_ogc_internal_minor_chemistry_wells_id " + "ON ogc_internal_minor_chemistry_wells (id)" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_water_elevation_wells") + op.execute(text(_create_water_elevation_view(public_only=False))) + op.execute( + text( + "COMMENT ON MATERIALIZED VIEW ogc_internal_water_elevation_wells IS " + "'Unfiltered latest water elevation per well with explicit units: " + "elevation_m, depth_to_water_below_ground_surface_ft, water_elevation_ft, for the internal pygeoapi mount.'" + ) + ) + op.execute( + text( + "CREATE UNIQUE INDEX ux_ogc_internal_water_elevation_wells_id " + "ON ogc_internal_water_elevation_wells (id)" + ) + ) + + # Recreate now that ogc_internal_water_well_summary exists again. + op.execute(text(_create_actively_monitored_wells_view())) + op.execute( + text( + "COMMENT ON VIEW ogc_internal_actively_monitored_wells IS " + "'Unfiltered wells in the Water Level Network group, for the internal pygeoapi mount.'" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_project_areas") + op.execute(text(_create_project_areas_view(public_only=False))) + op.execute( + text( + "COMMENT ON VIEW ogc_internal_project_areas IS " + "'Unfiltered project areas for groups with polygon boundaries, for the internal pygeoapi mount.'" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_locations") + op.execute(text(_create_locations_view())) + op.execute( + text( + "COMMENT ON VIEW ogc_internal_locations IS " + "'Unfiltered locations for the internal pygeoapi mount.'" + ) + ) + + +# All 22 relations this migration creates, in an order safe for DROP (the +# dependent view first, mirroring _recreate_all_internal_views's ordering). +ALL_INTERNAL_RELATIONS = [ + "ogc_internal_actively_monitored_wells", + *[f"ogc_internal_{view_id}" for view_id, _ in THING_VIEWS], + "ogc_internal_latest_depth_to_water_wells", + "ogc_internal_avg_tds_wells", + "ogc_internal_latest_tds_wells", + "ogc_internal_depth_to_water_trend_wells", + "ogc_internal_water_well_summary", + "ogc_internal_major_chemistry_results", + "ogc_internal_minor_chemistry_wells", + "ogc_internal_water_elevation_wells", + "ogc_internal_project_areas", + "ogc_internal_locations", +] + + +def upgrade() -> None: + _check_required_tables() + _recreate_all_internal_views() + + +def downgrade() -> None: + # None of these 22 relations existed before this migration -- unlike + # f4a5b6c7d8e9's downgrade (which recreates the prior unfiltered public + # views), there is no prior state to restore, so downgrade just drops + # everything this migration created. + for relation in ALL_INTERNAL_RELATIONS: + _drop_view_or_materialized_view(relation) diff --git a/alembic/versions/f4a5b6c7d8e9_apply_public_release_status_filter_to_ogc_views.py b/alembic/versions/f4a5b6c7d8e9_apply_public_release_status_filter_to_ogc_views.py index 3cc644393..eb403749d 100644 --- a/alembic/versions/f4a5b6c7d8e9_apply_public_release_status_filter_to_ogc_views.py +++ b/alembic/versions/f4a5b6c7d8e9_apply_public_release_status_filter_to_ogc_views.py @@ -468,6 +468,9 @@ def _create_water_well_summary_view(public_only: bool) -> str: # Static analyte columns for major chemistry pivots. # Includes aliases observed in current DB values (e.g., Ca(total), IONBAL, TAn, TCat, Na+K). +# Mirrored character-for-character (modulo view name) in +# 2d3c3a268652_create_internal_ogc_views.py; tests/test_migration_view_parity.py +# enforces the two stay in sync. STATIC_ANALYTE_COLUMNS_MAJOR: list[tuple[str, str]] = [ ("tds", "tds"), ("calcium", "calcium"), @@ -694,6 +697,9 @@ def _create_major_chemistry_results_view(public_only: bool) -> str: """ +# Mirrored character-for-character (modulo view name) in +# 2d3c3a268652_create_internal_ogc_views.py; tests/test_migration_view_parity.py +# enforces the two stay in sync. STATIC_ANALYTE_COLUMNS_MINOR: list[tuple[str, str]] = [ ("h2r", "h2r"), ("o18r", "o18r"), From 8eae5b17c70aa0ed432f5e4cf41f94b25bcf7ffd Mon Sep 17 00:00:00 2001 From: Kelsey Smuczynski Date: Fri, 17 Jul 2026 23:21:51 -0600 Subject: [PATCH 4/7] test(ogc): guard against analyte-mapping drift The major/minor chemistry CASE blocks now exist in two files (f4a5b6c7d8e9 and 2d3c3a268652) because migrations can't import from each other. Without a check, a future analyte-mapping fix applied to the public view (e.g. reported against a public-facing bug) has no reason to also touch the internal view sitting in a different file from a different ticket -- silently leaving internal staff looking at the stale mapping, which inverts who has the more reliable data. Compares AST source segments rather than raw text/regex so the comparison survives incidental formatting differences and only fails on a genuine logic change. Normalizes the one expected difference (the view name embedded in the CREATE statement) before comparing. --- tests/test_migration_view_parity.py | 76 +++++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 tests/test_migration_view_parity.py diff --git a/tests/test_migration_view_parity.py b/tests/test_migration_view_parity.py new file mode 100644 index 000000000..5c6035b80 --- /dev/null +++ b/tests/test_migration_view_parity.py @@ -0,0 +1,76 @@ +"""Drift detection between the public (A1) and internal (A11) OGC migrations. + +The two chemistry pivot views (major/minor) are dominated by long +analyte-alias CASE-mapping blocks that encode real lab-data business +knowledge. Because this codebase keeps Alembic migrations self-contained +with no cross-migration imports, that logic is duplicated rather than +shared between f4a5b6c7d8e9 (public) and 2d3c3a268652 (internal). If someone +fixes an analyte mapping in one file without the other, the public and +internal chemistry layers silently diverge -- this test turns that into a +loud, specific CI failure instead. +""" + +import ast +from pathlib import Path + +import pytest + +VERSIONS_DIR = Path(__file__).resolve().parent.parent / "alembic" / "versions" +PUBLIC_MIGRATION = ( + VERSIONS_DIR / "f4a5b6c7d8e9_apply_public_release_status_filter_to_ogc_views.py" +) +INTERNAL_MIGRATION = VERSIONS_DIR / "2d3c3a268652_create_internal_ogc_views.py" + +# Substrings that are expected to differ between the two files -- normalized +# away before comparison. Order matters: the internal_ variant must be +# stripped before its shorter public counterpart could ever match it. +NAME_SUBSTITUTIONS = [ + ("ogc_internal_major_chemistry_results", "ogc_major_chemistry_results"), + ("ogc_internal_minor_chemistry_wells", "ogc_minor_chemistry_wells"), +] + +COMPARED_NAMES = [ + "STATIC_ANALYTE_COLUMNS_MAJOR", + "STATIC_ANALYTE_COLUMNS_MINOR", + "_major_chemistry_select_columns", + "_major_chemistry_unit_columns", + "_minor_chemistry_value_columns", + "_minor_chemistry_unit_columns", + "_create_major_chemistry_results_view", + "_create_minor_chemistry_wells_view", +] + + +def _get_node_source(path: Path, name: str) -> str: + source = path.read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name == name: + return ast.get_source_segment(source, node) + if isinstance(node, ast.Assign): + targets = [t.id for t in node.targets if isinstance(t, ast.Name)] + if name in targets: + return ast.get_source_segment(source, node) + # STATIC_ANALYTE_COLUMNS_MAJOR/MINOR are annotated assignments + # (`NAME: list[...] = [...]`), which parse as AnnAssign, not Assign. + if isinstance(node, ast.AnnAssign): + if isinstance(node.target, ast.Name) and node.target.id == name: + return ast.get_source_segment(source, node) + raise AssertionError(f"{name!r} not found in {path}") + + +def _normalize(source: str) -> str: + for internal, public in NAME_SUBSTITUTIONS: + source = source.replace(internal, public) + return source + + +@pytest.mark.parametrize("name", COMPARED_NAMES) +def test_analyte_mapping_matches_between_public_and_internal_migrations(name): + public_source = _normalize(_get_node_source(PUBLIC_MIGRATION, name)) + internal_source = _normalize(_get_node_source(INTERNAL_MIGRATION, name)) + assert public_source == internal_source, ( + f"{name} has drifted between the public (f4a5b6c7d8e9) and internal " + "(2d3c3a268652) OGC migrations -- if this is a genuine analyte-mapping " + "fix, apply it to both files." + ) From f59a12ad0239a7687e0a8cd32db9d00ca9524682 Mon Sep 17 00:00:00 2001 From: Kelsey Smuczynski Date: Fri, 17 Jul 2026 23:22:49 -0600 Subject: [PATCH 5/7] test(ogc): add behave coverage for internal OGC mount Tags the 6 A11 scenarios A1 already wrote into this shared feature file with @production, appended to their existing tag lines rather than at the feature level -- this file has no feature-level tag since it's shared across ~10 other tickets' scenarios, and tagging at that level would pull in every other ticket's undefined steps. The 401/403/200 auth scenarios neutralize the ambient AUTHENTIK_DISABLE_AUTHENTICATION dev-bypass for their own duration only (restored via context.add_cleanup) and patch core.permissions.decode_token_payload directly, since no real Authentik server is available in CI to issue a genuine JWT. Existing auth-testing infrastructure in this repo only reaches app.dependency_overrides, which has zero effect on ASGI middleware -- none of it exercises InternalOGCAuthMiddleware for free. --- tests/features/ogc-cleanup-sprint1.feature | 12 +- tests/features/steps/ogc-cleanup-sprint1.py | 191 +++++++++++++++++++- 2 files changed, 193 insertions(+), 10 deletions(-) diff --git a/tests/features/ogc-cleanup-sprint1.feature b/tests/features/ogc-cleanup-sprint1.feature index c510bb0fc..c8c325dad 100644 --- a/tests/features/ogc-cleanup-sprint1.feature +++ b/tests/features/ogc-cleanup-sprint1.feature @@ -178,38 +178,38 @@ Feature: OGC Feature Layer Cleanup — Sprint 1 # A11 — Stand up authenticated internal OGC mount at /ogcapi-internal # --------------------------------------------------------------------------- - @backend @ogc-infrastructure @sprint-1 @high-priority @A11 + @backend @ogc-infrastructure @sprint-1 @high-priority @A11 @production Scenario: Anonymous request to internal OGC endpoint is rejected When an unauthenticated client requests /ogcapi-internal/collections Then the response HTTP status is 401 - @backend @ogc-infrastructure @sprint-1 @high-priority @A11 + @backend @ogc-infrastructure @sprint-1 @high-priority @A11 @production Scenario: Request with insufficient role to internal OGC endpoint is rejected Given the client presents a valid token with role "public-viewer" When the client requests /ogcapi-internal/collections Then the response HTTP status is 403 - @backend @ogc-infrastructure @sprint-1 @high-priority @A11 + @backend @ogc-infrastructure @sprint-1 @high-priority @A11 @production Scenario: Authenticated internal staff can access /ogcapi-internal collections Given an internal staff member with the required role is authenticated via Authentik When the staff member requests /ogcapi-internal/collections Then the response HTTP status is 200 And the response includes collections not available on the public /ogcapi endpoint - @backend @ogc-infrastructure @sprint-1 @high-priority @A11 + @backend @ogc-infrastructure @sprint-1 @high-priority @A11 @production Scenario: Internal collections expose private and draft records Given an authenticated internal staff member When the staff member requests items from the "water_wells" internal collection Then records with a release_status other than "public" are included in the response - @backend @ogc-infrastructure @sprint-1 @high-priority @A11 + @backend @ogc-infrastructure @sprint-1 @high-priority @A11 @production Scenario: Internal database relations are separate from public relations Given the /ogcapi-internal mount has been deployed When the database schema is inspected Then the database schema contains relations prefixed with "ogc_internal_" And no ogc_internal_ relation is shared with the public /ogcapi endpoint - @backend @ogc-infrastructure @sprint-1 @high-priority @A11 + @backend @ogc-infrastructure @sprint-1 @high-priority @A11 @production Scenario: Public /ogcapi surface is unaffected by the internal mount When a client requests /ogcapi/collections Then no collection in the response has an id prefixed "ogc_internal_" diff --git a/tests/features/steps/ogc-cleanup-sprint1.py b/tests/features/steps/ogc-cleanup-sprint1.py index 5c2f3e5ee..51e744d8a 100644 --- a/tests/features/steps/ogc-cleanup-sprint1.py +++ b/tests/features/steps/ogc-cleanup-sprint1.py @@ -13,15 +13,18 @@ # See the License for the specific language governing permissions and # limitations under the License. # =============================================================================== -"""Step definitions for A1 (public release_status filter on ogc_* views). +"""Step definitions for A1 (public release_status filter on ogc_* views) and +A11 (authenticated internal OGC mount at /ogcapi-internal). -Only the @A1-tagged scenarios in ogc-cleanup-sprint1.feature are implemented -here. The other ~10 tickets sharing that feature file have no steps yet and -stay undefined/dormant, per this ticket's plan. +Only the @A1- and @A11-tagged scenarios in ogc-cleanup-sprint1.feature are +implemented here. The other ~9 tickets sharing that feature file have no +steps yet and stay undefined/dormant, per this ticket's plan. """ import importlib +import os from datetime import date +from unittest.mock import patch from alembic import command from behave import given, when, then @@ -34,6 +37,7 @@ admin_function, amp_admin_function, ) +from core.permissions import INTERNAL_OGC_GROUP from starlette.testclient import TestClient from db import ( @@ -670,4 +674,183 @@ def step_then_same_feature_count_as_before(context): ), f"{layer_id}: expected {before_count} features (unchanged), got {after_count}" +# --------------------------------------------------------------------------- +# A11 -- authenticated internal OGC mount (/ogcapi-internal) +# --------------------------------------------------------------------------- +# +# tests/test_pygeoapi_mount.py's existing coverage and the "a functioning +# api" step above both only touch FastAPI's app.dependency_overrides, which +# has zero effect on ASGI middleware -- none of this codebase's existing +# auth-testing infrastructure reaches InternalOGCAuthMiddleware for free. +# The 401/403/200 scenarios below instead neutralize the ambient +# AUTHENTIK_DISABLE_AUTHENTICATION dev-bypass (set for the whole bdd-tests CI +# job) for the scenario's duration only, and control +# core.permissions.decode_token_payload's return value directly, since no +# real Authentik server is available in CI to issue a genuine JWT. + + +def _neutralize_authentik_bypass(context): + """Temporarily force the dev-bypass off so InternalOGCAuthMiddleware's + real 401/403 logic actually runs, restored unconditionally afterward so + later scenarios relying on the global CI dev-bypass are unaffected. + """ + original = os.environ.get("AUTHENTIK_DISABLE_AUTHENTICATION") + os.environ["AUTHENTIK_DISABLE_AUTHENTICATION"] = "0" + + def _restore(): + if original is None: + os.environ.pop("AUTHENTIK_DISABLE_AUTHENTICATION", None) + else: + os.environ["AUTHENTIK_DISABLE_AUTHENTICATION"] = original + + context.add_cleanup(_restore) + + +def _patch_decode_token_payload(context, groups): + patcher = patch( + "core.permissions.decode_token_payload", return_value={"groups": groups} + ) + patcher.start() + context.add_cleanup(patcher.stop) + + +def _teardown_a11_seed_data(): + with session_ctx() as session: + session.execute(text("DELETE FROM thing WHERE name LIKE 'A11 %'")) + session.commit() + + +@when("an unauthenticated client requests /ogcapi-internal/collections") +def step_when_unauthenticated_client_requests_internal_collections(context): + _neutralize_authentik_bypass(context) + context.response = context.client.get("/ogcapi-internal/collections") + + +@given('the client presents a valid token with role "{role}"') +def step_given_client_presents_token_with_role(context, role): + _neutralize_authentik_bypass(context) + _patch_decode_token_payload(context, groups=[role]) + context.auth_token = "a11-behave-test-token" + + +@when("the client requests /ogcapi-internal/collections") +def step_when_client_requests_internal_collections(context): + headers = {"Authorization": f"Bearer {context.auth_token}"} + context.response = context.client.get( + "/ogcapi-internal/collections", headers=headers + ) + + +@given("an internal staff member with the required role is authenticated via Authentik") +def step_given_internal_staff_authenticated_via_authentik(context): + _neutralize_authentik_bypass(context) + _patch_decode_token_payload(context, groups=[INTERNAL_OGC_GROUP]) + context.auth_token = "a11-behave-test-token" + + +@when("the staff member requests /ogcapi-internal/collections") +def step_when_staff_member_requests_internal_collections(context): + headers = {"Authorization": f"Bearer {context.auth_token}"} + context.response = context.client.get( + "/ogcapi-internal/collections", headers=headers + ) + + +@then("the response includes collections not available on the public /ogcapi endpoint") +def step_then_response_includes_internal_only_collections(context): + payload = context.response.json() + collections = payload.get("collections", []) + assert collections, "internal endpoint returned no collections" + for collection in collections: + links = collection.get("links", []) + assert any("/ogcapi-internal" in link.get("href", "") for link in links), ( + f"{collection.get('id')}: no self-link referencing /ogcapi-internal -- " + "expected each internal collection representation to be reachable " + "only via the internal mount, not the public /ogcapi endpoint" + ) + + +@given("an authenticated internal staff member") +def step_given_an_authenticated_internal_staff_member(context): + # Relies on the ambient AUTHENTIK_DISABLE_AUTHENTICATION dev-bypass (set + # for the whole CI job) rather than a real token -- this scenario is + # about what the internal mount exposes, not auth semantics (covered + # separately by the 401/403/200 scenarios above). + with session_ctx() as session: + context.a11_seed_ids = {} + for status in STATUSES: + thing = _seed_thing_with_location( + session, "water well", status, f"A11 {status}" + ) + context.a11_seed_ids[status] = thing.id + context.add_cleanup(_teardown_a11_seed_data) + + +@when('the staff member requests items from the "{layer_id}" internal collection') +def step_when_staff_member_requests_internal_collection_items(context, layer_id): + context.response = context.client.get( + f"/ogcapi-internal/collections/{layer_id}/items?limit=500" + ) + + +@then('records with a release_status other than "public" are included in the response') +def step_then_non_public_records_included(context): + payload = context.response.json() + ids_present = _layer_feature_ids(payload) + non_public_ids = {context.a11_seed_ids["private"], context.a11_seed_ids["draft"]} + assert non_public_ids & ids_present, ( + "expected the internal collection to include the seeded private/draft " + f"wells {non_public_ids}, got ids {ids_present}" + ) + + +@given("the /ogcapi-internal mount has been deployed") +def step_given_internal_mount_has_been_deployed(context): + command.upgrade(_alembic_config(), "head") + + +@when("the database schema is inspected") +def step_when_database_schema_is_inspected(context): + with session_ctx() as session: + context.schema_relations = set( + session.execute( + text( + "SELECT c.relname FROM pg_class c " + "JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE c.relkind IN ('v', 'm') AND n.nspname = 'public'" + ) + ).scalars() + ) + + +@then('the database schema contains relations prefixed with "{prefix}"') +def step_then_schema_contains_relations_prefixed(context, prefix): + matching = {r for r in context.schema_relations if r.startswith(prefix)} + assert matching, f"expected at least one relation prefixed {prefix!r}, found none" + + +@then("no ogc_internal_ relation is shared with the public /ogcapi endpoint") +def step_then_no_internal_relation_shared_with_public(context): + internal = {r for r in context.schema_relations if r.startswith("ogc_internal_")} + assert internal, "no ogc_internal_ relations found in the schema" + for relation in internal: + public_equivalent = relation.replace("ogc_internal_", "ogc_", 1) + assert public_equivalent in context.schema_relations, ( + f"{relation} has no distinct public counterpart ({public_equivalent}) in " + "the schema -- expected the two sets to coexist as separate relations" + ) + + +@when("a client requests /ogcapi/collections") +def step_when_client_requests_ogcapi_collections(context): + context.response = context.client.get("/ogcapi/collections") + + +@then('no collection in the response has an id prefixed "{prefix}"') +def step_then_no_collection_id_prefixed(context, prefix): + payload = context.response.json() + offending = [c["id"] for c in payload["collections"] if c["id"].startswith(prefix)] + assert not offending, f"found collections with id prefixed {prefix!r}: {offending}" + + # ============= EOF ============================================= From 3882a5df77138a46e34fd53711284892e01a2408 Mon Sep 17 00:00:00 2001 From: Kelsey Smuczynski Date: Mon, 3 Aug 2026 11:43:38 -0600 Subject: [PATCH 6/7] fix(ogc): drop duplicate collections step definition Merging A1's updated branch (itself synced with a newer staging) brought in tests/features/steps/edr_water_data.py, which defines its own "a client requests /ogcapi/collections" step. Behave raises AmbiguousStep when identical step text is registered twice across files, so this step no longer defines it here, leaving the functionally identical one in edr_water_data.py as the sole definition. --- tests/features/steps/ogc-cleanup-sprint1.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/features/steps/ogc-cleanup-sprint1.py b/tests/features/steps/ogc-cleanup-sprint1.py index 51e744d8a..2b27e539f 100644 --- a/tests/features/steps/ogc-cleanup-sprint1.py +++ b/tests/features/steps/ogc-cleanup-sprint1.py @@ -841,9 +841,11 @@ def step_then_no_internal_relation_shared_with_public(context): ) -@when("a client requests /ogcapi/collections") -def step_when_client_requests_ogcapi_collections(context): - context.response = context.client.get("/ogcapi/collections") +# "a client requests /ogcapi/collections" is defined in +# tests/features/steps/edr_water_data.py (functionally identical: GETs +# /ogcapi/collections and stashes context.response) -- reused rather than +# redefined here, since Behave raises AmbiguousStep on duplicate step text +# across files. @then('no collection in the response has an id prefixed "{prefix}"') From cc5f72cfec7318b9200cc2fd9c013a972ee4aa23 Mon Sep 17 00:00:00 2001 From: Kelsey Smuczynski Date: Mon, 3 Aug 2026 11:45:06 -0600 Subject: [PATCH 7/7] feat(ogc): mirror EDR views in internal OGC mount The merge that brought A1 up to date with staging also added ogc_waterlevels/ogc_water_chemistry (ADR3's EDR feature), which did not exist when A11's plan scoped this migration to 22 relations. Given the choice to leave them out or extend to match, chose to extend: A11's purpose is giving staff unfiltered access to what the public mount filters, and these two are filtered the same way as the original 22. ogc_internal_waterlevels/ogc_internal_water_chemistry mirror the public views with all three release_status predicates dropped (both the manual-reading and chemistry selects, plus the transducer union). _edr_collections_block now takes table_prefix, threaded through the same include_edr flag that keeps EDR off the internal mount if it is ever wired wrong again. Added a forward-looking parity test that diffs the public and internal relation sets directly, rather than checking a hardcoded count. It would have caught this exact gap, a new public view landing with no internal mirror, on its own instead of needing a human to notice. --- .../2d3c3a268652_create_internal_ogc_views.py | 134 +++++++++++++++++- core/pygeoapi.py | 20 ++- tests/test_migration_view_parity.py | 69 +++++++++ 3 files changed, 214 insertions(+), 9 deletions(-) diff --git a/alembic/versions/2d3c3a268652_create_internal_ogc_views.py b/alembic/versions/2d3c3a268652_create_internal_ogc_views.py index affda020d..a7e7c5835 100644 --- a/alembic/versions/2d3c3a268652_create_internal_ogc_views.py +++ b/alembic/versions/2d3c3a268652_create_internal_ogc_views.py @@ -6,6 +6,14 @@ (core/pygeoapi.py::mount_pygeoapi_internal). Full parity with the public set, per ticket A11 -- not a subset. +Also mirrors ogc_waterlevels/ogc_water_chemistry from z9a0b1c2d3e4 (added to +staging after this migration's original 22-relation scope was written, per +ADR3's EDR feature) as ogc_internal_waterlevels/ogc_internal_water_chemistry, +bringing the total to 24. Unfiltered in all three places the public views +predicate on release_status: the manual-readings and chemistry selects' +`o.release_status = 'public'`, and the transducer union's +`tobs.release_status = 'public'`. + The major/minor chemistry analyte-mapping CASE blocks and STATIC_ANALYTE_COLUMNS lists below are intentionally character-for-character identical (modulo view name) to their counterparts in f4a5b6c7d8e9 -- this @@ -25,7 +33,7 @@ of that JOIN, it must be dropped before ogc_internal_water_well_summary and recreated after (same ordering constraint as the public side). -All 22 relations here are newly created by this migration -- none of them +All 24 relations here are newly created by this migration -- none of them existed in any form beforehand -- so downgrade() simply drops them rather than recreating a prior state. @@ -61,6 +69,11 @@ "NMA_MajorChemistry", "NMA_Chemistry_SampleInfo", "NMA_MinorTraceChemistry", + # For the ogc_internal_waterlevels/ogc_internal_water_chemistry EDR + # mirrors (see z9a0b1c2d3e4_add_edr_water_views.py). + "transducer_observation", + "deployment", + "parameter", } LATEST_LOCATION_CTE = """ @@ -1118,6 +1131,101 @@ def _create_locations_view() -> str: """ +# Shared join from a thing to its current location point -- same shape as +# z9a0b1c2d3e4's _LOCATION_JOIN. +_EDR_LOCATION_JOIN = """ + JOIN location_thing_association lta + ON lta.thing_id = t.id AND lta.effective_end IS NULL + JOIN location l ON l.id = lta.location_id +""" + + +def _create_internal_waterlevels_view() -> str: + # Mirrors z9a0b1c2d3e4's ogc_waterlevels with both release_status + # predicates dropped (manual readings: o.release_status; transducer + # readings: tobs.release_status). release_status itself is still + # selected as a column, same as the public view. + return f""" + CREATE VIEW ogc_internal_waterlevels AS + -- manual water-level readings + SELECT + 'm-' || o.id AS id, + t.id AS thing_id, + t.name AS station_name, + ST_X(l.point) AS longitude, + ST_Y(l.point) AS latitude, + o.observation_datetime AS datetime, + o.value AS value, + o.unit AS unit, + 'groundwater level' AS parameter_name, + 'manual' AS source, + NULL::integer AS deployment_id, + o.release_status AS release_status + FROM observation o + JOIN parameter p + ON p.id = o.parameter_id AND p.parameter_name = 'groundwater level' + JOIN sample sm ON sm.id = o.sample_id + JOIN field_activity fa ON fa.id = sm.field_activity_id + JOIN field_event fe ON fe.id = fa.field_event_id + JOIN thing t ON t.id = fe.thing_id + {_EDR_LOCATION_JOIN} + WHERE o.value IS NOT NULL + + UNION ALL + + -- transducer (instrument) water-level readings + SELECT + 't-' || tobs.id AS id, + t.id AS thing_id, + t.name AS station_name, + ST_X(l.point) AS longitude, + ST_Y(l.point) AS latitude, + tobs.observation_datetime AS datetime, + tobs.value AS value, + p.default_unit AS unit, + 'groundwater level' AS parameter_name, + 'transducer' AS source, + tobs.deployment_id AS deployment_id, + tobs.release_status AS release_status + FROM transducer_observation tobs + JOIN parameter p + ON p.id = tobs.parameter_id AND p.parameter_name = 'groundwater level' + JOIN deployment d ON d.id = tobs.deployment_id + JOIN thing t ON t.id = d.thing_id + {_EDR_LOCATION_JOIN} + WHERE tobs.value IS NOT NULL + """ + + +def _create_internal_water_chemistry_view() -> str: + # Mirrors z9a0b1c2d3e4's ogc_water_chemistry with its release_status + # predicate (o.release_status) dropped. + return f""" + CREATE VIEW ogc_internal_water_chemistry AS + SELECT + 'c-' || o.id AS id, + t.id AS thing_id, + t.name AS station_name, + ST_X(l.point) AS longitude, + ST_Y(l.point) AS latitude, + o.observation_datetime AS datetime, + o.value AS value, + o.unit AS unit, + p.parameter_name AS parameter_name, + o.sample_id AS sample_id, + o.release_status AS release_status + FROM observation o + JOIN parameter p + ON p.id = o.parameter_id AND p.parameter_name <> 'groundwater level' + JOIN sample sm ON sm.id = o.sample_id + JOIN field_activity fa ON fa.id = sm.field_activity_id + JOIN field_event fe ON fe.id = fa.field_event_id + JOIN thing t ON t.id = fe.thing_id + {_EDR_LOCATION_JOIN} + WHERE o.value IS NOT NULL + """ + + def _recreate_all_internal_views() -> None: # ogc_internal_actively_monitored_wells depends on # ogc_internal_water_well_summary via a direct JOIN; Postgres refuses to @@ -1271,8 +1379,26 @@ def _recreate_all_internal_views() -> None: ) ) + _drop_view_or_materialized_view("ogc_internal_waterlevels") + op.execute(text(_create_internal_waterlevels_view())) + op.execute( + text( + "COMMENT ON VIEW ogc_internal_waterlevels IS " + "'Unfiltered depth-to-water readings (manual + transducer) for the internal pygeoapi mount.'" + ) + ) + + _drop_view_or_materialized_view("ogc_internal_water_chemistry") + op.execute(text(_create_internal_water_chemistry_view())) + op.execute( + text( + "COMMENT ON VIEW ogc_internal_water_chemistry IS " + "'Unfiltered water-chemistry analyses (by analyte) for the internal pygeoapi mount.'" + ) + ) + -# All 22 relations this migration creates, in an order safe for DROP (the +# All 24 relations this migration creates, in an order safe for DROP (the # dependent view first, mirroring _recreate_all_internal_views's ordering). ALL_INTERNAL_RELATIONS = [ "ogc_internal_actively_monitored_wells", @@ -1287,6 +1413,8 @@ def _recreate_all_internal_views() -> None: "ogc_internal_water_elevation_wells", "ogc_internal_project_areas", "ogc_internal_locations", + "ogc_internal_waterlevels", + "ogc_internal_water_chemistry", ] @@ -1296,7 +1424,7 @@ def upgrade() -> None: def downgrade() -> None: - # None of these 22 relations existed before this migration -- unlike + # None of these 24 relations existed before this migration -- unlike # f4a5b6c7d8e9's downgrade (which recreates the prior unfiltered public # views), there is no prior state to restore, so downgrade just drops # everything this migration created. diff --git a/core/pygeoapi.py b/core/pygeoapi.py index 2bea76daa..5d6503011 100644 --- a/core/pygeoapi.py +++ b/core/pygeoapi.py @@ -260,9 +260,15 @@ def _edr_collections_block( dbname: str, user: str, password_placeholder: str, + table_prefix: str = "ogc_", ) -> str: resources: dict[str, dict] = {} for collection in EDR_COLLECTIONS: + # EDR_COLLECTIONS' table values are hardcoded to the public + # ogc_waterlevels/ogc_water_chemistry names -- strip that literal + # "ogc_" so table_prefix (here, "ogc_internal_" for the internal + # mount) still applies, the same way _thing_collections_block does. + table_name = table_prefix + collection["table"].removeprefix("ogc_") provider = { "type": "edr", "name": "core.edr_provider.WaterEDRProvider", @@ -274,7 +280,7 @@ def _edr_collections_block( "password": password_placeholder, }, "id_field": "id", - "table": collection["table"], + "table": table_name, } if collection["instance_field"]: provider["instance_field"] = collection["instance_field"] @@ -354,11 +360,11 @@ def _write_config( table_prefix=table_prefix, ) if include_edr: - # EDR collections (core/edr_provider.py) are backed by ogc_waterlevels/ - # ogc_water_chemistry, which are always public-filtered (see - # z9a0b1c2d3e4_add_edr_water_views.py) with no table_prefix - # parametrization -- there is no internal, unfiltered counterpart, so - # this only ever applies to the public mount's config. + # EDR collections (core/edr_provider.py), backed by + # ogc_waterlevels/ogc_water_chemistry (public) or + # ogc_internal_waterlevels/ogc_internal_water_chemistry (internal, + # see 2d3c3a268652_create_internal_ogc_views.py) depending on + # table_prefix. thing_collections_block = "\n".join( [ thing_collections_block, @@ -368,6 +374,7 @@ def _write_config( dbname=dbname, user=user, password_placeholder=password_placeholder, + table_prefix=table_prefix, ), ] ) @@ -495,6 +502,7 @@ def mount_pygeoapi_internal(app: FastAPI) -> None: server_url=_internal_server_url(), table_prefix="ogc_internal_", template_path=_internal_template_path(), + include_edr=True, ) _generate_openapi(config_path, openapi_path) _assert_server_settings_match(_pygeoapi_dir() / "pygeoapi-config.yml", config_path) diff --git a/tests/test_migration_view_parity.py b/tests/test_migration_view_parity.py index 5c6035b80..65e109b1e 100644 --- a/tests/test_migration_view_parity.py +++ b/tests/test_migration_view_parity.py @@ -11,6 +11,7 @@ """ import ast +import re from pathlib import Path import pytest @@ -20,6 +21,7 @@ VERSIONS_DIR / "f4a5b6c7d8e9_apply_public_release_status_filter_to_ogc_views.py" ) INTERNAL_MIGRATION = VERSIONS_DIR / "2d3c3a268652_create_internal_ogc_views.py" +EDR_MIGRATION = VERSIONS_DIR / "z9a0b1c2d3e4_add_edr_water_views.py" # Substrings that are expected to differ between the two files -- normalized # away before comparison. Order matters: the internal_ variant must be @@ -74,3 +76,70 @@ def test_analyte_mapping_matches_between_public_and_internal_migrations(name): "(2d3c3a268652) OGC migrations -- if this is a genuine analyte-mapping " "fix, apply it to both files." ) + + +# --------------------------------------------------------------------------- +# Forward-looking coverage check: does every public ogc_* relation (across +# f4a5b6c7d8e9 and any later migration that adds more, like z9a0b1c2d3e4's +# EDR views) have an ogc_internal_ mirror? This is exactly the gap that let +# the EDR views land on staging with no internal counterpart in the first +# place -- nothing caught it until a human noticed. +# --------------------------------------------------------------------------- + +# Matches "CREATE VIEW ogc_x AS" / "CREATE MATERIALIZED VIEW ogc_internal_x +# AS" for any *literal* relation name. Deliberately does not match the 11 +# thing-type views on either side of the parity: their names are built from +# an f-string variable (ogc_{safe_view_id}), not a literal, in both files -- +# handled separately via THING_VIEWS below. +_CREATE_VIEW_RE = re.compile( + r"CREATE (?:MATERIALIZED )?VIEW (ogc_(?:internal_)?[A-Za-z0-9_]+) AS" +) + + +def _literal_view_names(path: Path) -> set[str]: + return set(_CREATE_VIEW_RE.findall(path.read_text(encoding="utf-8"))) + + +def _thing_view_ids(path: Path) -> set[str]: + source = path.read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + targets = [t.id for t in node.targets if isinstance(t, ast.Name)] + if "THING_VIEWS" in targets and isinstance(node.value, ast.List): + return { + elt.elts[0].value + for elt in node.value.elts + if isinstance(elt, ast.Tuple) + } + raise AssertionError(f"THING_VIEWS not found in {path}") + + +def test_internal_migration_mirrors_every_public_relation(): + public_thing_ids = _thing_view_ids(PUBLIC_MIGRATION) + internal_thing_ids = _thing_view_ids(INTERNAL_MIGRATION) + assert public_thing_ids == internal_thing_ids, ( + f"THING_VIEWS has drifted: public has {public_thing_ids}, " + f"internal has {internal_thing_ids}" + ) + + public_relation_ids = { + name.removeprefix("ogc_") + for name in _literal_view_names(PUBLIC_MIGRATION) + | _literal_view_names(EDR_MIGRATION) + } | public_thing_ids + internal_relation_ids = { + name.removeprefix("ogc_internal_") + for name in _literal_view_names(INTERNAL_MIGRATION) + } | internal_thing_ids + + missing = public_relation_ids - internal_relation_ids + assert not missing, ( + "public ogc_* relations with no ogc_internal_ mirror in " + f"{INTERNAL_MIGRATION.name}: {sorted(missing)}" + ) + assert len(public_relation_ids) == 24, ( + "expected 24 total relations (11 thing-type + 11 from f4a5b6c7d8e9 + " + f"2 EDR from z9a0b1c2d3e4), got {len(public_relation_ids)}: " + f"{sorted(public_relation_ids)}" + )