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 =============================================