From 7aa174631546ae1673f7c293baf3d92039ae08f9 Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 24 Aug 2026 11:07:02 -0700 Subject: [PATCH] feat(ogc): expose last_observation_date on the Group A layers The 11 thing-type layers carried construction and location detail but no signal of data recency: nothing distinguished a well measured last month from one last visited in 1994 without querying a second layer. This adds last_observation_date to the shared thing-view template -- the UTC date of the most recent observation recorded against the thing, or NULL where it has none -- and rebuilds all 11 public views and their ogc_internal_ counterparts from that one template so both mounts stay identical. Scope is the observation table, reached through the sample -> field_activity -> field_event chain every other observation-backed view here uses. Continuous transducer readings are deliberately excluded: they hang off deployments rather than field events, they exist for a handful of instrumented wells rather than for Group A generally, and a max() over the largest table in the schema would need an index of its own to stay cheap. Wells with logger data are served by actively_monitored_wells and the water-elevation layers. Per-thing lookup is a LEFT JOIN LATERAL so a paginated or single-feature request touches only the observations of the rows it returns. That join path had no indexes at all, so the four it needs come with the migration. The behave harness needed one fix to go with this: pygeoapi caches reflected table models process-wide, so the scenarios that downgrade the schema under a running app were building SELECTs naming a column the downgraded views no longer have. The cache is now cleared wherever those scenarios move the schema. Co-Authored-By: Claude Opus 5 --- ...dd_last_observation_date_to_thing_views.py | 239 ++++++++++++++++++ core/ogc-field-descriptions.yml | 9 + tests/features/environment.py | 16 ++ tests/features/ogc-cleanup-sprint1.feature | 6 +- tests/features/steps/ogc-cleanup-sprint1.py | 234 ++++++++++++++++- 5 files changed, 494 insertions(+), 10 deletions(-) create mode 100644 alembic/versions/b8c9d0e1f2a3_add_last_observation_date_to_thing_views.py diff --git a/alembic/versions/b8c9d0e1f2a3_add_last_observation_date_to_thing_views.py b/alembic/versions/b8c9d0e1f2a3_add_last_observation_date_to_thing_views.py new file mode 100644 index 00000000..0c04fb2b --- /dev/null +++ b/alembic/versions/b8c9d0e1f2a3_add_last_observation_date_to_thing_views.py @@ -0,0 +1,239 @@ +"""add last_observation_date to the Group A thing views + +Ticket A13. The 11 thing-type layers (Group A) carry construction and location +detail but no signal of data recency: a consumer could not tell a well measured +last month from one last visited in 1994 without querying a second layer. + +This adds `last_observation_date` to the shared thing-view template -- the date +of the most recent observation recorded against the thing, or NULL where the +thing has no observations at all. All 11 public views and their 11 +`ogc_internal_` counterparts are rebuilt from the same template here, so the +two mounts stay column-for-column identical. + +Scope of "observation": rows in the `observation` table, reached through the +sample -> field_activity -> field_event chain that every other observation- +backed view in this schema uses. Continuous transducer readings +(`transducer_observation`) are deliberately *not* folded in: they live on a +different chain (deployment -> thing), they exist for a handful of instrumented +water wells rather than for Group A generally, and a max() over the largest +table in the schema would need its own index on +(deployment_id, observation_datetime) to stay cheap. Wells with logger data are +served by ogc_actively_monitored_wells and the water-elevation layers. If +Group A currency should later include instrument readings, that is a separate +ticket and a separate index. + +The date is the UTC calendar date of the observation timestamp -- same +convention as transducer_daily_data (v0w1x2y3z4a5) -- rather than a +session-timezone cast, so the value does not depend on who is querying. + +Public views count only observations with release_status='public', matching how +the public mount filters everything else; the internal views count all of them. +A public well whose only observations are private therefore reads NULL on +/ogcapi and carries a date on /ogcapi-internal. + +Per-thing lookup is a LEFT JOIN LATERAL rather than a grouped CTE so that a +paginated or single-feature request touches only the observations of the rows +it returns. That path had no indexes at all (Postgres does not index foreign +keys on its own), so the four it needs are created here. + +The view bodies below are otherwise character-for-character the templates from +f4a5b6c7d8e9 (public) and 2d3c3a268652 (internal); downgrade() restores them. + +Revision ID: b8c9d0e1f2a3 +Revises: 986e0eb85ab3 +Create Date: 2026-08-24 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 = "b8c9d0e1f2a3" +down_revision: Union[str, Sequence[str], None] = "baba91fe5e83" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +REQUIRED_TABLES = { + "thing", + "location", + "location_thing_association", + "observation", + "sample", + "field_activity", + "field_event", +} + +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"), +] + +# (name, table, columns) for the observation chain the lateral walks +# thing -> field_event -> field_activity -> sample -> observation. +SUPPORTING_INDEXES = [ + ("ix_field_event_thing_id", "field_event", "thing_id"), + ("ix_field_activity_field_event_id", "field_activity", "field_event_id"), + ("ix_sample_field_activity_id", "sample", "field_activity_id"), + ( + "ix_observation_sample_id_observation_datetime", + "observation", + "sample_id, observation_datetime", + ), +] + + +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 _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 add last_observation_date to the OGC thing views. " + f"Missing required tables: {', '.join(sorted(missing))}" + ) + + +def _create_thing_view( + view_id: str, thing_type: str, public_only: bool, table_prefix: str +) -> str: + """The Group A view template, with last_observation_date.""" + safe_view_id = _safe_view_id(f"{table_prefix}{view_id}") + escaped_thing_type = thing_type.replace("'", "''") + release_filter = " AND t.release_status = 'public'" if public_only else "" + observation_release_filter = ( + "\n AND o.release_status = 'public'" if public_only else "" + ) + return f""" + CREATE VIEW {safe_view_id} AS + WITH latest_location AS ( +{LATEST_LOCATION_CTE} + ) + SELECT + t.id, + t.name, + t.first_visit_date, + ( + last_obs.last_observation_datetime AT TIME ZONE 'UTC' + )::date AS last_observation_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 + LEFT JOIN LATERAL ( + SELECT MAX(o.observation_datetime) AS last_observation_datetime + 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 + WHERE fe.thing_id = t.id{observation_release_filter} + ) AS last_obs ON TRUE + WHERE t.thing_type = '{escaped_thing_type}'{release_filter} + """ + + +def _create_thing_view_pre_a13( + view_id: str, thing_type: str, public_only: bool, table_prefix: str +) -> str: + """The template as it stood in f4a5b6c7d8e9/2d3c3a268652, for downgrade.""" + safe_view_id = _safe_view_id(f"{table_prefix}{view_id}") + escaped_thing_type = thing_type.replace("'", "''") + release_filter = " AND t.release_status = 'public'" if public_only else "" + return f""" + CREATE VIEW {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 _rebuild_thing_views(builder) -> None: + for table_prefix, public_only in (("ogc_", True), ("ogc_internal_", False)): + for view_id, thing_type in THING_VIEWS: + view_name = _safe_view_id(f"{table_prefix}{view_id}") + op.execute(text(f"DROP VIEW IF EXISTS {view_name}")) + op.execute(text(builder(view_id, thing_type, public_only, table_prefix))) + + +def upgrade() -> None: + _check_required_tables() + + for index_name, table_name, columns in SUPPORTING_INDEXES: + op.execute( + text(f"CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} ({columns})") + ) + + _rebuild_thing_views(_create_thing_view) + + +def downgrade() -> None: + _rebuild_thing_views(_create_thing_view_pre_a13) + + for index_name, _table_name, _columns in SUPPORTING_INDEXES: + op.execute(text(f"DROP INDEX IF EXISTS {index_name}")) diff --git a/core/ogc-field-descriptions.yml b/core/ogc-field-descriptions.yml index dd01ac95..90cf606e 100644 --- a/core/ogc-field-descriptions.yml +++ b/core/ogc-field-descriptions.yml @@ -40,6 +40,15 @@ _defaults: first_visit_date: title: First visit date description: Date of the earliest Bureau visit on record for this feature. + last_observation_date: + title: Last observation date + description: >- + Date of the most recent measurement recorded against this feature, as a + UTC calendar date. Null where no measurement is on record for it. Counts + readings and laboratory results held in the observation record; continuous + instrument readings from a deployed logger are not included, so an + instrumented well can carry newer data than this date shows. On the public + mount only measurements released to the public are counted. nma_pk_welldata: title: Legacy NM_Aquifer well key description: >- diff --git a/tests/features/environment.py b/tests/features/environment.py index 2a7af12d..d3c1b47c 100644 --- a/tests/features/environment.py +++ b/tests/features/environment.py @@ -645,6 +645,20 @@ def _alembic_config() -> Config: return cfg +def reset_pygeoapi_reflection() -> None: + """Drop pygeoapi's process-wide cache of reflected table models. + + pygeoapi.provider.sql.get_table_model is functools.cache'd, so a provider + keeps serving the column list it reflected the first time a collection was + queried. Scenarios that move the schema under a running app (the + @migration-mutates-schema ones) would otherwise build SELECTs naming + columns the downgraded views no longer have. + """ + from pygeoapi.provider.sql import get_table_model + + get_table_model.cache_clear() + + def _initialize_test_schema() -> None: with session_ctx() as session: recreate_public_schema(session) @@ -876,6 +890,7 @@ def before_scenario(context, scenario): # Defense in depth against a previous, unrelated failure having # already left the database below head. command.upgrade(_alembic_config(), "head") + reset_pygeoapi_reflection() def after_scenario(context, scenario): @@ -885,6 +900,7 @@ def after_scenario(context, scenario): # this database. Deliberately not gated on DROP_AND_REBUILD_DB, # since these scenarios mutate schema regardless of that flag. command.upgrade(_alembic_config(), "head") + reset_pygeoapi_reflection() if not get_bool_env("DROP_AND_REBUILD_DB"): return diff --git a/tests/features/ogc-cleanup-sprint1.feature b/tests/features/ogc-cleanup-sprint1.feature index bfab8bc9..d6a823e9 100644 --- a/tests/features/ogc-cleanup-sprint1.feature +++ b/tests/features/ogc-cleanup-sprint1.feature @@ -203,7 +203,7 @@ Feature: OGC Feature Layer Cleanup — Sprint 1 # A13 — Add last_observation_date column to Group A view template # --------------------------------------------------------------------------- - @backend @ogc-data-currency @sprint-1 @medium-priority @A13 + @backend @ogc-data-currency @sprint-1 @medium-priority @A13 @production Scenario: last_observation_date column is present in all Group A layers When a client requests items from each of the following layers: | layer-id | @@ -221,7 +221,7 @@ Feature: OGC Feature Layer Cleanup — Sprint 1 # other_things is not listed: it is in the Group A view template, but A18 # took it off the public catalog — it is only reachable on /ogcapi-internal. - @backend @ogc-data-currency @sprint-1 @medium-priority @A13 + @backend @ogc-data-currency @sprint-1 @medium-priority @A13 @production Scenario: last_observation_date is NULL for things with no associated observations Given monitoring locations with no linked observations exist in each of the following layers: | layer-id | @@ -240,7 +240,7 @@ Feature: OGC Feature Layer Cleanup — Sprint 1 # other_things is not listed: it is in the Group A view template, but A18 # took it off the public catalog — it is only reachable on /ogcapi-internal. - @backend @ogc-data-currency @sprint-1 @medium-priority @A13 + @backend @ogc-data-currency @sprint-1 @medium-priority @A13 @production Scenario: Consumers can filter Group A layers by last_observation_date Given each of the following Group A layers has features with last_observation_date values "2019-06-01" and "2023-06-01": | layer-id | diff --git a/tests/features/steps/ogc-cleanup-sprint1.py b/tests/features/steps/ogc-cleanup-sprint1.py index 8497cca1..5ccf0b22 100644 --- a/tests/features/steps/ogc-cleanup-sprint1.py +++ b/tests/features/steps/ogc-cleanup-sprint1.py @@ -14,12 +14,14 @@ # limitations under the License. # =============================================================================== """Step definitions for A1 (public release_status filter on ogc_* views), -A2 (OGC server metadata placeholders) and A11 (authenticated internal OGC -mount at /ogcapi-internal). - -Only the @A1-, @A2- and @A11-tagged scenarios in ogc-cleanup-sprint1.feature -are implemented here. The other ~8 tickets sharing that feature file have no -steps yet and stay undefined/dormant, per those tickets' plans. +A2 (OGC server metadata placeholders), A11 (authenticated internal OGC +mount at /ogcapi-internal) and A13 (last_observation_date on the Group A +view template). + +Only the @A1-, @A2-, @A11- and @A13-tagged scenarios in +ogc-cleanup-sprint1.feature are implemented here. The other tickets sharing +that feature file have no steps yet and stay undefined/dormant, per those +tickets' plans. """ import importlib @@ -62,7 +64,7 @@ ) from db.engine import session_ctx from tests import get_parameter_id -from tests.features.environment import _alembic_config +from tests.features.environment import _alembic_config, reset_pygeoapi_reflection # Revision immediately before this ticket's schema migration -- re-verify # with `alembic heads`/`alembic history` if this file is revisited later, @@ -649,6 +651,7 @@ def _seed_already_consistent_layers(session): @given("the following layers were already filtering correctly before the migration:") def step_given_already_consistent_layers(context): command.downgrade(_alembic_config(), PRE_A1_REVISION) + reset_pygeoapi_reflection() with session_ctx() as session: _seed_already_consistent_layers(session) session.commit() @@ -664,6 +667,7 @@ def step_given_already_consistent_layers(context): @when("the Sprint 1 migration is applied") def step_when_sprint1_migration_is_applied(context): command.upgrade(_alembic_config(), "head") + reset_pygeoapi_reflection() @then("each of those layers returns the same feature count as before the migration") @@ -1023,4 +1027,220 @@ def step_then_terms_of_service_resolves(context): ), f"{terms_url!r} resolved but does not look like the disclaimer page" +# --------------------------------------------------------------------------- +# A13 -- last_observation_date on the Group A view template +# --------------------------------------------------------------------------- + +# Every Group A layer the A13 scenarios name, plus water_wells, which is not in +# SIMPLE_THING_TYPE_LAYERS because the A1 scenarios seed it with a full +# observation/chemistry chain rather than a bare Location + Thing. +A13_LAYER_THING_TYPES = {"water_wells": "water well", **dict(SIMPLE_THING_TYPE_LAYERS)} + +# Dates the filter scenario splits on: one comfortably before its 2021-01-01 +# cutoff, one comfortably after. +A13_STALE_DATE = "2019-06-01" +A13_RECENT_DATE = "2023-06-01" + + +def _seed_thing_with_observation(session, thing_type, name, observation_date=None): + """A public thing of `thing_type`, optionally with one public observation. + + `observation_date` is a plain YYYY-MM-DD string; it is stored at midday UTC + so the view's UTC-date cast cannot land on the neighbouring day. + """ + thing = _seed_thing_with_location(session, thing_type, "public", name) + if observation_date is None: + return thing + + field_event = FieldEvent( + thing_id=thing.id, + event_date=f"{observation_date}T12:00:00Z", + notes="A13 behave seed field event", + release_status="public", + ) + session.add(field_event) + session.commit() + + field_activity = FieldActivity( + field_event_id=field_event.id, + activity_type="groundwater level", + notes="A13 behave seed field activity", + release_status="public", + ) + session.add(field_activity) + session.commit() + + sample = Sample( + field_activity_id=field_activity.id, + sample_date=f"{observation_date}T12:00:00Z", + sample_name=f"A13 sample {thing.id}", + sample_matrix="water", + sample_method="Steel-tape measurement", + qc_type="Normal", + notes="A13 behave seed sample", + release_status="public", + ) + session.add(sample) + session.commit() + + observation = Observation( + observation_datetime=f"{observation_date}T12:00:00Z", + sample_id=sample.id, + parameter_id=get_parameter_id("groundwater level", "Field Parameter"), + release_status="public", + value=12.0, + unit="ft", + measuring_point_height=1.0, + groundwater_level_reason="Water level not affected", + ) + session.add(observation) + session.commit() + + return thing + + +def _get_item(context, layer_id, feature_id): + response = context.client.get(f"/ogcapi/collections/{layer_id}/items/{feature_id}") + assert response.status_code == 200, ( + f"Unexpected status {response.status_code} for {layer_id}/{feature_id}: " + f"{response.text}" + ) + return response.json() + + +@when("a client requests items from each of the following layers:") +def step_when_client_requests_items_from_layers(context): + context.layer_responses = {} + for row in context.table: + layer_id = row["layer-id"].strip() + context.layer_responses[layer_id] = _get_items(context, layer_id) + + +@then("each feature includes a last_observation_date property") +def step_then_each_feature_includes_last_observation_date(context): + for layer_id, payload in context.layer_responses.items(): + # Several Group A thing types carry no seeded rows in the behave + # database, and an empty feature list would let a missing column pass + # unnoticed -- so the layer's own queryables are checked as well. + queryables = context.client.get(f"/ogcapi/collections/{layer_id}/queryables") + assert queryables.status_code == 200, ( + f"queryables for {layer_id} returned {queryables.status_code}: " + f"{queryables.text}" + ) + advertised = queryables.json().get("properties", {}) + assert "last_observation_date" in advertised, ( + f"{layer_id} does not advertise last_observation_date: " + f"{sorted(advertised)}" + ) + + for feature in payload["features"]: + assert "last_observation_date" in feature["properties"], ( + f"{layer_id} feature {feature.get('id')} has no " + f"last_observation_date property: {sorted(feature['properties'])}" + ) + + +@given( + "monitoring locations with no linked observations exist in each of the following layers:" +) +def step_given_things_without_observations(context): + context.a13_unobserved_ids = {} + with session_ctx() as session: + for row in context.table: + layer_id = row["layer-id"].strip() + thing_type = A13_LAYER_THING_TYPES[layer_id] + thing = _seed_thing_with_observation( + session, thing_type, f"A13 unobserved {layer_id}" + ) + context.a13_unobserved_ids[layer_id] = thing.id + + +@when("a client requests those features") +def step_when_client_requests_those_features(context): + context.a13_unobserved_features = { + layer_id: _get_item(context, layer_id, feature_id) + for layer_id, feature_id in context.a13_unobserved_ids.items() + } + + +@then("each feature's last_observation_date property is null") +def step_then_last_observation_date_is_null(context): + for layer_id, feature in context.a13_unobserved_features.items(): + value = feature["properties"]["last_observation_date"] + assert value is None, ( + f"{layer_id} feature {feature.get('id')} has last_observation_date " + f"{value!r}; a thing with no observations must read null" + ) + + +@given( + "each of the following Group A layers has features with last_observation_date " + 'values "{stale_date}" and "{recent_date}":' +) +def step_given_layers_with_stale_and_recent_observations( + context, stale_date, recent_date +): + context.a13_stale_ids = {} + context.a13_recent_ids = {} + with session_ctx() as session: + for row in context.table: + layer_id = row["layer-id"].strip() + thing_type = A13_LAYER_THING_TYPES[layer_id] + stale = _seed_thing_with_observation( + session, thing_type, f"A13 stale {layer_id}", stale_date + ) + recent = _seed_thing_with_observation( + session, thing_type, f"A13 recent {layer_id}", recent_date + ) + context.a13_stale_ids[layer_id] = stale.id + context.a13_recent_ids[layer_id] = recent.id + + +@when("a client requests items from each of those layers with filter") +def step_when_client_requests_layers_with_filter(context): + cql = context.text.strip() + context.layer_responses = {} + for layer_id in context.a13_recent_ids: + response = context.client.get( + f"/ogcapi/collections/{layer_id}/items", + params={"filter": cql, "filter-lang": "cql2-text", "limit": 200}, + ) + assert response.status_code == 200, ( + f"Filtered request on {layer_id} returned {response.status_code}: " + f"{response.text}" + ) + context.layer_responses[layer_id] = response.json() + + +@then( + 'only features with a last_observation_date of "{recent_date}" are returned ' + "from each layer" +) +def step_then_only_recent_features_returned(context, recent_date): + cutoff = date.fromisoformat("2021-01-01") + for layer_id, payload in context.layer_responses.items(): + returned_ids = _layer_feature_ids(payload) + recent_id = context.a13_recent_ids[layer_id] + stale_id = context.a13_stale_ids[layer_id] + + assert ( + recent_id in returned_ids + ), f"{layer_id} dropped its {recent_date} feature (id={recent_id})" + assert stale_id not in returned_ids, ( + f"{layer_id} returned its {A13_STALE_DATE} feature (id={stale_id}) " + "through a filter that excludes it" + ) + + for feature in payload["features"]: + value = feature["properties"]["last_observation_date"] + assert value is not None, ( + f"{layer_id} feature {feature.get('id')} passed the filter with " + "a null last_observation_date" + ) + assert date.fromisoformat(value[:10]) > cutoff, ( + f"{layer_id} feature {feature.get('id')} has last_observation_date " + f"{value!r}, which the filter should have excluded" + ) + + # ============= EOF =============================================