diff --git a/.env.example b/.env.example index bfa04542d..2e2bd4556 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/alembic/versions/2d3c3a268652_create_internal_ogc_views.py b/alembic/versions/2d3c3a268652_create_internal_ogc_views.py new file mode 100644 index 000000000..a7e7c5835 --- /dev/null +++ b/alembic/versions/2d3c3a268652_create_internal_ogc_views.py @@ -0,0 +1,1432 @@ +"""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. + +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 +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 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. + +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", + # 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 = """ +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 + """ + + +# 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 + # 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.'" + ) + ) + + _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 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", + *[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", + "ogc_internal_waterlevels", + "ogc_internal_water_chemistry", +] + + +def upgrade() -> None: + _check_required_tables() + _recreate_all_internal_views() + + +def downgrade() -> None: + # 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. + 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 c5587e22f..82659be52 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 @@ -476,6 +476,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"), @@ -702,6 +705,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"), 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/factory.py b/core/factory.py index 3877e7bf3..f85b6f4cf 100644 --- a/core/factory.py +++ b/core/factory.py @@ -42,9 +42,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) configure_cors_middleware(app) configure_apitally_middleware(app) return app 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 ============================================= 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 0cd69672a..5d6503011 100644 --- a/core/pygeoapi.py +++ b/core/pygeoapi.py @@ -137,13 +137,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("/"): @@ -156,21 +160,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: @@ -178,10 +188,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 @@ -192,6 +211,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: @@ -219,7 +239,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", } ], @@ -240,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", @@ -254,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"] @@ -315,41 +341,58 @@ 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, + include_edr: bool = False, +) -> None: host, port, dbname, user, password_placeholder = _pygeoapi_db_settings() - template = _template_path().read_text(encoding="utf-8") - config = template.format( - server_url=_server_url(), - postgres_host=host, - postgres_port=port, - postgres_db=dbname, - postgres_user=user, - postgres_password_env=password_placeholder, - thing_collections_block="\n".join( + template = (template_path or _template_path()).read_text(encoding="utf-8") + thing_collections_block = _thing_collections_block( + host=host, + port=port, + dbname=dbname, + user=user, + password_placeholder=password_placeholder, + table_prefix=table_prefix, + ) + if include_edr: + # 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( - host=host, - port=port, - dbname=dbname, - user=user, - password_placeholder=password_placeholder, - ), + thing_collections_block, _edr_collections_block( host=host, port=port, dbname=dbname, user=user, password_placeholder=password_placeholder, + table_prefix=table_prefix, ), ] - ), + ) + config = template.format( + server_url=server_url, + postgres_host=host, + postgres_port=port, + postgres_db=dbname, + postgres_user=user, + postgres_password_env=password_placeholder, + thing_collections_block=thing_collections_block, ) - # 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. @@ -357,6 +400,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 @@ -389,7 +458,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(), include_edr=True) _generate_openapi(config_path, openapi_path) os.environ["PYGEOAPI_CONFIG"] = str(config_path) @@ -400,3 +469,52 @@ 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(), + include_edr=True, + ) + _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 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..2b27e539f 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,185 @@ 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" + ) + + +# "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}"') +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 ============================================= diff --git a/tests/test_migration_view_parity.py b/tests/test_migration_view_parity.py new file mode 100644 index 000000000..65e109b1e --- /dev/null +++ b/tests/test_migration_view_parity.py @@ -0,0 +1,145 @@ +"""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 +import re +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" +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 +# 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." + ) + + +# --------------------------------------------------------------------------- +# 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)}" + )