From 63bf5023e1cd4e538671350b698acbd4107d5562 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 15:42:01 -0700 Subject: [PATCH 1/3] feat(transducer): publish and range-delete for corrected hydrographs The hydrograph corrector in OcotilloUI could only download its corrected series as CSV -- there was no POST for transducer observations at all. This adds the two write endpoints its upload contract specifies. POST /observation/transducer-groundwater-level/block publishes one corrected logger file as one block plus all of its readings, in one transaction. The block's span is derived from the measurements rather than sent: nothing links the observation table to the block table, so the reader pairs them by time and a client-supplied span wider than the data would make the block claim readings it does not contain. `deployment_id` is optional and resolved from the deployments covering that span; zero or more than one match is a 422 rather than a guess, because guessing attributes readings to hardware that did not record them. An existing block sharing any instant with the new one is a 409 listing the collisions. `?replace_overlapping=true` deletes those blocks and their readings. The readings have to go with the block -- keeping them would leave rows the reader cannot show that still occupy the deployment/parameter/instant the new series is about to claim, so a "replace" that kept them would fail on the very insert it was asked to make room for. Readings orphaned by a hand-deleted block are caught separately and reported with the earliest colliding timestamp, rather than letting the insert abort on a constraint name. Overlap is inclusive on both bounds, unlike TransducerObservationBlock.overlaps, which is half-open. The reader matches with `start <= t <= end`, so two blocks sharing an endpoint both claim a reading at that instant -- exactly the ambiguity the check exists to prevent. DELETE /observation/transducer-groundwater-level removes every reading for a well inside a closed range and reconciles the blocks that covered them: one left empty is deleted, one left partial has its span narrowed to the survivors. All three parameters are required; there is deliberately no unbounded form. Scope matches the GET on the same path, so the set a client previews is the set this removes. Schema changes: provenance on the block (source_file, source_kind, and an ordered corrections list), and a per-reading `note` set only where a correction moved the value, so NULL reads as "as measured" rather than "unknown". A corrected block is derived data, and a reviewer who cannot see that a series was snapped to a manual measurement cannot review it. The block time-order check is relaxed to `end >= start`: a block covering a single instant is legitimate, either published that way or narrowed to it by a delete. Both write routes are gated on AMP.Staging, a standalone group -- AMPAdmin does not satisfy it and it satisfies nothing else -- so they ship dark while the workbench is validated against real logger files. Two bugs fixed in passing: - The read route called get_transducer_observations positionally, and the helper's fourth positional parameter is `sensor_id`. `start_time` landed in `sensor_id` (unused, dropped), `end_time` landed in `start_time`, and `end_time` was never set, so a requested upper bound was ignored and the lower bound came from the wrong argument. - `sort`/`order` on that route were accepted and ignored. They now work over a whitelist; an unrecognised field is a 422 rather than a silently differently ordered page. Wellntel support from the contract is deferred: both the readings proxy and the sensor_type filter on /thing are blocked on where the API key and the wellname/PointID mapping should live. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 6 + ...4e5f6a7b8_hydrograph_correction_publish.py | 116 ++++ api/observation.py | 118 +++- core/dependencies.py | 17 + db/transducer.py | 45 +- docs/hydrograph-correction-publish.md | 110 ++++ domain/hydrograph.py | 163 +++++ schemas/transducer.py | 130 +++- services/observation_helper.py | 49 +- services/transducer_helper.py | 459 +++++++++++++ tests/test_authorization.py | 1 + tests/test_domain_hydrograph.py | 179 +++++ tests/test_transducer_publish.py | 610 ++++++++++++++++++ 13 files changed, 1988 insertions(+), 15 deletions(-) create mode 100644 alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py create mode 100644 docs/hydrograph-correction-publish.md create mode 100644 domain/hydrograph.py create mode 100644 services/transducer_helper.py create mode 100644 tests/test_domain_hydrograph.py create mode 100644 tests/test_transducer_publish.py diff --git a/CLAUDE.md b/CLAUDE.md index 30549235f..1bb5bdcaa 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -164,6 +164,12 @@ Authentik groups granted. **Role families are orthogonal**: general `Admin` confers nothing in the AMP or Lexicon families. Only tiers *within* a family nest. +**`AMP.Staging`** is a standalone group, not a fourth AMP tier — `AMPAdmin` +does not satisfy it. It gates the hydrograph corrector's publish and range-delete +routes while the workbench is being validated against real logger files, so they +ship dark. Read **`docs/hydrograph-correction-publish.md`** before changing +them. + **Authorization is opt-in per endpoint** — a `user: _dependency` parameter in the signature, not a router-level `dependencies=[...]`. Omitting it produces a fully public endpoint with no error. `tests/test_authorization.py` holds the diff --git a/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py b/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py new file mode 100644 index 000000000..d3f20e363 --- /dev/null +++ b/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py @@ -0,0 +1,116 @@ +"""publish provenance for corrected transducer blocks + +Revision ID: c3d4e5f6a7b8 +Revises: b2c3d4e5f6a7 +Create Date: 2026-08-19 + +The hydrograph corrector publishes a *derived* series: water head converted to +depth below ground surface against manual anchors, then shifted, snapped, and +drift-corrected. None of those numbers are what the instrument recorded, so the +database has to carry enough to tell a reviewer what happened to them. + +Three columns on the block cover the batch: the file it came from, whether that +file held water head or depth to water, and the ordered list of corrections +applied. `comment` already exists and takes the publisher's free-text note. + +One column on the observation covers the row: `note`, set only on readings a +correction actually moved. NULL therefore means "as measured", which is the +distinction review needs. The legacy `nma_waterlevelscontinuous_*_notes` +columns cannot serve -- each is scoped to one legacy source table. + +The block time-order check is relaxed from `>` to `>=`. A block spanning a +single instant is legitimate: a published file with one reading, or a block +narrowed by a range delete until one observation survives. The block reader +matches observations inclusively on both bounds, so a zero-width block still +covers its reading. Loosening a check constraint cannot invalidate existing +rows. +""" + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision = "c3d4e5f6a7b8" +down_revision = "b2c3d4e5f6a7" +branch_labels = None +depends_on = None + +# Spelled as it exists in the database, typo included -- renaming it here would +# leave deployed environments with a constraint this migration cannot find. +TIME_ORDER_CONSTRAINT = "check_transuder_block_time_order" + + +def upgrade() -> None: + op.add_column( + "transducer_observation_block", + sa.Column( + "source_file", + sa.String(length=255), + nullable=True, + comment="Name of the logger file the corrected series was derived from", + ), + ) + op.add_column( + "transducer_observation_block", + sa.Column( + "source_kind", + sa.String(length=50), + nullable=True, + comment="What the source file measured: water_head or depth_to_water", + ), + ) + op.add_column( + "transducer_observation_block", + sa.Column( + "corrections", + postgresql.JSONB(astext_type=sa.Text()), + nullable=True, + comment="Corrections applied to the source series, in applied order", + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "note", + sa.Text(), + nullable=True, + comment=( + "Per-reading correction annotation; NULL means the value is as " + "measured" + ), + ), + ) + + op.drop_constraint( + TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check" + ) + op.create_check_constraint( + TIME_ORDER_CONSTRAINT, + "transducer_observation_block", + "end_datetime >= start_datetime", + ) + + +def downgrade() -> None: + # Zero-width blocks may have been created while the loosened constraint was + # in force, so widen them by a second rather than let the stricter + # constraint fail to validate. A one-second span on a block that covered an + # instant is a smaller lie than a failed downgrade. + op.execute( + "UPDATE transducer_observation_block " + "SET end_datetime = start_datetime + interval '1 second' " + "WHERE end_datetime = start_datetime" + ) + op.drop_constraint( + TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check" + ) + op.create_check_constraint( + TIME_ORDER_CONSTRAINT, + "transducer_observation_block", + "end_datetime > start_datetime", + ) + + op.drop_column("transducer_observation", "note") + op.drop_column("transducer_observation_block", "corrections") + op.drop_column("transducer_observation_block", "source_kind") + op.drop_column("transducer_observation_block", "source_file") diff --git a/api/observation.py b/api/observation.py index d4c7fff78..4e084036c 100644 --- a/api/observation.py +++ b/api/observation.py @@ -28,6 +28,7 @@ session_dependency, amp_admin_dependency, amp_editor_dependency, + amp_staging_dependency, amp_viewer_dependency, ) from db import Observation, Parameter @@ -40,7 +41,12 @@ UpdateGroundwaterLevelObservation, UpdateWaterChemistryObservation, ) -from schemas.transducer import TransducerObservationWithBlockResponse +from schemas.transducer import ( + DeletedTransducerObservationsResponse, + PublishedTransducerBlockResponse, + PublishTransducerBlock, + TransducerObservationWithBlockResponse, +) from schemas.water_level_csv import WaterLevelBulkUploadResponse from services.crud_helper import model_deleter, model_adder from services.observation_helper import ( @@ -50,10 +56,30 @@ get_transducer_observations, ) from services.query_helper import simple_get_by_id +from services.transducer_helper import ( + delete_transducer_observations, + publish_transducer_block, +) from services.water_level_csv import bulk_upload_water_levels router = APIRouter(prefix="/observation", tags=["observation"]) + +def _groundwater_level_parameter_id(session) -> int: + """ + The lexicon id the transducer routes work in. + + Looked up rather than configured so the publish, read, and delete routes + cannot drift onto different parameters. + """ + return ( + session.query(Parameter) + .filter(Parameter.parameter_name == "groundwater level") + .one() + .id + ) + + """ TODO @@ -88,6 +114,33 @@ def add_water_chemistry_observation( return model_adder(session, Observation, obs_data, user=user) +@router.post( + "/transducer-groundwater-level/block", + status_code=HTTP_201_CREATED, + summary="Publish a corrected transducer series as one block", +) +def publish_transducer_groundwater_level_block( + payload: PublishTransducerBlock, + session: session_dependency, + user: amp_staging_dependency, + replace_overlapping: bool = False, +) -> PublishedTransducerBlockResponse: + """ + Publish one corrected logger file as a single observation block. + + The block's time span is derived from the measurements; the client does not + send it. Overlapping an existing block is a 409 listing the collisions -- + pass `replace_overlapping=true` to supersede them, which deletes those + blocks and their readings in the same transaction. + + Written by the hydrograph corrector in OcotilloUI. See + `docs/hydrograph-correction-publish.md`. + """ + return publish_transducer_block( + session, payload, user=user, replace_overlapping=replace_overlapping + ) + + @router.post( "/groundwater-level/bulk-upload", response_model=WaterLevelBulkUploadResponse, @@ -155,17 +208,30 @@ def get_transducer_groundwater_level_observations( thing_id: int | None = None, start_time: datetime | None = None, end_time: datetime | None = None, + sort: str | None = None, + order: str | None = None, ) -> CustomPage[TransducerObservationWithBlockResponse]: + """ + Retrieve transducer groundwater level observations paired with the block + that covers them. - groundwater_parameter_id = ( - session.query(Parameter) - .filter(Parameter.parameter_name == "groundwater level") - .one() - .id - ) - + `sort` accepts `observation_datetime`, `value`, or `id`; `order` accepts + `asc` or `desc`. The default is newest first, so a client that wants the + latest stored reading for a well can ask for size 1. + """ + # Keyword arguments deliberately: the helper's fourth positional parameter + # is `sensor_id`, so the previous positional call passed `start_time` as a + # sensor id (unused, silently dropped), `end_time` as `start_time`, and + # nothing as `end_time` -- an upper bound the caller asked for was ignored + # and the lower bound came from the wrong argument. return get_transducer_observations( - session, thing_id, groundwater_parameter_id, start_time, end_time + session, + thing_id=thing_id, + parameter_id=_groundwater_level_parameter_id(session), + start_time=start_time, + end_time=end_time, + sort=sort, + order=order, ) @@ -302,6 +368,40 @@ def get_observation_by_id( # DELETE ======================================================================= +@router.delete( + "/transducer-groundwater-level", + status_code=HTTP_200_OK, + summary="Delete transducer groundwater level observations in a time range", +) +def delete_transducer_groundwater_level_observations( + session: session_dependency, + user: amp_staging_dependency, + thing_id: int, + start_time: datetime, + end_time: datetime, +) -> DeletedTransducerObservationsResponse: + """ + Delete every transducer groundwater level reading for a well inside a + closed time range, and reconcile the blocks that covered them: a block left + with no readings is deleted, one left with some has its span narrowed to + the survivors. + + All three parameters are required -- there is deliberately no form of this + request that deletes everything for a well. Scoped exactly like the `GET` + on this path, so the set previewed there is the set removed here. + + Irreversible, and it leaves the `transducer_daily_data` materialized view + stale until its next refresh. + """ + return delete_transducer_observations( + session, + thing_id=thing_id, + parameter_id=_groundwater_level_parameter_id(session), + start_time=start_time, + end_time=end_time, + ) + + @router.delete( "/{observation_id}", summary="Delete an observation", diff --git a/core/dependencies.py b/core/dependencies.py index 09e7c3f79..95d11f3c8 100644 --- a/core/dependencies.py +++ b/core/dependencies.py @@ -59,6 +59,21 @@ amp_viewer_function = authenticated(any_of=["AMPAdmin", "AMPEditor", "AMPViewer"]) +# Hydrograph-Corrector Staging Permissions ------------------------------------- +# The hydrograph corrector's publish and range-delete routes write and destroy +# transducer records, and the workbench driving them is still being validated +# against real logger files. `AMP.Staging` is its own group with no tier below +# it and no AMP tier above it -- an AMPAdmin does not satisfy it. Nobody holds +# it until it is granted in Authentik, so the routes ship dark and reachable +# only by whoever is testing them. +# +# This is deliberately not a fourth rung on the AMP ladder. When the workbench +# is trusted, these routes move to `amp_admin_dependency` and the group goes +# away; leaving it as a tier would make that a schema change instead of a +# one-line edit. +amp_staging_function = authenticated(any_of=["AMP.Staging"]) + + # Lexicon-Specific Authentication/Permissions ---------------------------------- lexicon_admin_function = authenticated(any_of=["LexiconAdmin"]) @@ -89,5 +104,7 @@ amp_editor_dependency: TypeAlias = Annotated[dict, Depends(amp_editor_function)] amp_viewer_dependency: TypeAlias = Annotated[dict, Depends(amp_viewer_function)] +amp_staging_dependency: TypeAlias = Annotated[dict, Depends(amp_staging_function)] + no_permission_dependency: TypeAlias = Annotated[dict, Depends(no_permission_function)] # ============= EOF ============================================= diff --git a/db/transducer.py b/db/transducer.py index d109adc58..65766f6e5 100644 --- a/db/transducer.py +++ b/db/transducer.py @@ -28,6 +28,7 @@ Index, UniqueConstraint, ) +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import mapped_column, Mapped, relationship from db import Base, AutoBaseMixin, ReleaseMixin, lexicon_term @@ -62,6 +63,32 @@ class TransducerObservationBlock(Base, AutoBaseMixin, ReleaseMixin): ) comment: Mapped[str] = mapped_column(Text, nullable=True) + + # Publish provenance. A corrected block is derived data -- the numbers in it + # are not what any instrument recorded -- so the file it came from and the + # operations applied to it are part of the record, not metadata about it. A + # reviewer who cannot see that a series was snapped to a manual measurement + # cannot review it. + source_file: Mapped[str] = mapped_column( + String(255), + nullable=True, + comment="Name of the logger file the corrected series was derived from", + ) + source_kind: Mapped[str] = mapped_column( + String(50), + nullable=True, + comment="What the source file measured: water_head or depth_to_water", + ) + # A list of strings in applied order rather than a modelled correction + # entity: the corrector's operation set is still moving, and freezing it + # into columns now would mean a migration per new operation. The strings + # are written by the workbench and read by humans. + corrections: Mapped[list] = mapped_column( + JSONB, + nullable=True, + comment="Corrections applied to the source series, in applied order", + ) + reviewer_id: Mapped[str] = mapped_column( ForeignKey("contact.id", ondelete="CASCADE"), nullable=True, @@ -81,8 +108,13 @@ class TransducerObservationBlock(Base, AutoBaseMixin, ReleaseMixin): "end_datetime", name="uq_transducer_block_thing_status_parameter_time", ), + # Non-strict: a block covering a single instant is legitimate -- a + # published file with one reading, or a block narrowed by a range + # delete until one observation survives. The block reader matches + # observations inclusively on both bounds, so a zero-width block still + # covers its reading. CheckConstraint( - "end_datetime > start_datetime", name="check_transuder_block_time_order" + "end_datetime >= start_datetime", name="check_transuder_block_time_order" ), Index( "ix_transducer_block_time", @@ -137,6 +169,17 @@ class TransducerObservation(Base, AutoBaseMixin, ReleaseMixin): ) value: Mapped[float] = mapped_column(Float, nullable=False) + # Why this reading differs from what the sensor recorded. Present only on + # readings a correction actually moved, so a NULL note means the value is + # as measured -- which is the distinction review needs and which the legacy + # `nma_waterlevelscontinuous_*_notes` columns cannot carry, being scoped to + # one legacy source each. + note: Mapped[str] = mapped_column( + Text, + nullable=True, + comment="Per-reading correction annotation; NULL means the value is as measured", + ) + # How far through review this reading is, on USGS terms: provisional, # in review, approved. Orthogonal to `release_status`, which says who may # see it -- a reading can be public and provisional at once, which one diff --git a/docs/hydrograph-correction-publish.md b/docs/hydrograph-correction-publish.md new file mode 100644 index 000000000..572e4485f --- /dev/null +++ b/docs/hydrograph-correction-publish.md @@ -0,0 +1,110 @@ +# Hydrograph correction — publish and range delete + +The hydrograph corrector in OcotilloUI (`/ocotillo/hydrograph-correction`) +ingests a raw logger file, converts water head to depth below ground surface +against manual measurements, applies corrections, and publishes the result +here. This document records what the API side actually does; the UI-side +proposal it was built from is +`OcotilloUI/docs/hydrograph-correction-upload-contract.md`. + +## Authorization + +Both write routes are gated on **`AMP.Staging`**, a standalone Authentik group. +It is not a fourth rung on the AMP ladder: `AMPAdmin` does not satisfy it, and +it satisfies nothing else. Nobody holds it until it is granted, so the routes +ship dark and are reachable only by whoever is validating the workbench against +real logger files. + +When the workbench is trusted, these routes move to `amp_admin_dependency` and +the group goes away. Leaving it as a tier would make that a schema change +instead of a one-line edit. + +The read route stays on `amp_viewer_dependency` — it was already public to +viewers and publishing does not change who may look. + +## `POST /observation/transducer-groundwater-level/block` + +One corrected logger file becomes one block plus all of its readings, in one +transaction. + +- **The span is derived, not sent.** `start_datetime`/`end_datetime` come from + the min/max measurement timestamp. A client-supplied span wider than the data + would make the block claim readings it does not contain, because nothing links + the observation table to the block table — the reader pairs them by time. +- **`deployment_id` is optional.** Omitted, it is resolved from the deployments + on the well whose installation period covers the span. A NULL installation date + reads as "always installed", a NULL removal date as "still installed". Zero or + more than one match is a 422 telling the client to send it explicitly, because + guessing attributes readings to hardware that did not record them. +- **`data_maturity` is derived from `review_status`**, not sent: a block + published as `not reviewed` is `provisional` on USGS terms. Sending both + separately would let a client store a contradiction. +- **Provenance is part of the record.** `source_file`, `source_kind`, and the + ordered `corrections` list live on the block; `provenance.notes` lands in the + block's existing `comment`. A reviewer who cannot see that a series was + snapped to a manual measurement cannot review it. +- **Per-reading `note`** is set only where a correction moved the value, so NULL + means "as measured" rather than "unknown". + +### Overlap + +An existing block for the same well and parameter whose span shares any instant +with the new one is a **409** listing the collisions in +`detail[0].input.overlapping_blocks`. `?replace_overlapping=true` deletes those +blocks **and their readings** in the same transaction and then publishes. + +The readings have to go with the block. Keeping them would leave rows the reader +cannot show — no block covers them — that still occupy the +deployment/parameter/instant the new series is about to claim, so a "replace" +that kept them would fail on the very insert it was asked to make room for. + +Overlap here is **inclusive on both bounds**, unlike +`TransducerObservationBlock.overlaps` on the model, which is half-open. The +reader matches a reading to a block with `start <= t <= end`, so two blocks +sharing an endpoint both claim any reading at that instant — exactly the +ambiguity this check exists to prevent. + +Readings can also survive a block deleted by hand. Those are caught separately +and reported as a 409 naming the earliest colliding timestamp, rather than +letting the insert abort the transaction with a constraint name. + +## `DELETE /observation/transducer-groundwater-level` + +`thing_id`, `start_time`, and `end_time` are all required. There is deliberately +no unbounded form of this request. The scope matches the `GET` on the same path +exactly, so the set a client previews is the set this removes. + +Blocks are reconciled afterwards: one left with no readings is deleted, one left +with some has its span narrowed to the survivors. A block narrowed to a single +reading becomes zero-width, which the `end_datetime >= start_datetime` check +constraint allows on purpose (migration `c3d4e5f6a7b8`) and which the inclusive +reader still covers. + +**This leaves the `transducer_daily_data` materialized view stale** until its +next scheduled refresh. Nothing here refreshes it — a full refresh on every +delete would cost far more than the correctness it buys between nightly runs. + +## Two things fixed in passing + +- The read route was calling `get_transducer_observations` positionally, and the + helper's fourth positional parameter is `sensor_id`. `start_time` was landing + in `sensor_id` (unused, silently dropped), `end_time` was landing in + `start_time`, and `end_time` was never set — so an upper bound a caller asked + for was ignored and the lower bound came from the wrong argument. The call is + keyword-only now. +- The read route honours `sort` (`observation_datetime`, `value`, `id`) and + `order` (`asc`/`desc`), defaulting to newest first. An unrecognised sort field + is a 422 rather than being ignored — silently returning a differently ordered + page reads as the data changing, not as a bad request. + +## Not built + +Everything in the contract's "Supporting endpoints for Wellntel ingestion" +section is deferred: the `GET /wellntel/readings` proxy and the `sensor_type` +filter on `GET /thing`. Both are blocked on open questions the contract itself +raises — where the Wellntel API key lives and where the wellname→PointID mapping +belongs. The UI already falls back to demo data when they are absent. + +Also open, and unchanged by this work: whether the raw water-head series should +be retained alongside the corrected one, and whether publishing as `provisional` +should feed a review queue. diff --git a/domain/hydrograph.py b/domain/hydrograph.py new file mode 100644 index 000000000..20ad1c7ed --- /dev/null +++ b/domain/hydrograph.py @@ -0,0 +1,163 @@ +# =============================================================================== +# Copyright 2026 ross +# +# 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. +# =============================================================================== +""" +Rules for publishing and deleting corrected transducer series. + +The hydrograph corrector (OcotilloUI) uploads a whole corrected file as one +block. These functions decide the block's span, whether it collides with what is +already stored, which deployment it belongs to, and what survives a range +delete. They take plain values -- no session, no request -- so the awkward parts +(inclusive vs half-open overlap, a block narrowed to a single instant) can be +tested without a database. + +**Overlap is inclusive on both bounds here**, which differs from +``TransducerObservationBlock.overlaps`` on the model. That method is half-open, +so two blocks sharing an endpoint do not "overlap". The block *reader* +(``services.observation_helper.get_transducer_observations``) matches an +observation to a block with ``start <= t <= end``, so two blocks sharing an +endpoint both claim any reading at that instant and the reader picks whichever +sorts first. That ambiguity is the thing the publish conflict check exists to +prevent, so the check uses the reader's inclusive bounds rather than the +model's. +""" + +from datetime import date, datetime + +# One request per logger file is the expected shape; a 90-day file at a 6-hour +# cadence is 360 rows. The cap is three orders of magnitude above that, high +# enough never to reject real work and low enough that a runaway client cannot +# ask the server to build a million-row transaction. +MAX_MEASUREMENTS = 100_000 + + +class HydrographError(ValueError): + """Base for publish/delete rule violations. A ValueError, per ADR4.""" + + +def derive_block_span( + observation_datetimes: list[datetime], +) -> tuple[datetime, datetime]: + """ + The block's span is the extent of its readings. + + The client does not send ``start_datetime``/``end_datetime``: a span wider + than the data would claim coverage the block does not have, and the reader + would attach unrelated readings to it. + """ + if not observation_datetimes: + raise HydrographError("A block needs at least one measurement") + + return min(observation_datetimes), max(observation_datetimes) + + +def first_out_of_order_index(observation_datetimes: list[datetime]) -> int | None: + """ + Index of the first timestamp that does not advance on its predecessor. + + Strictly increasing, so a repeated timestamp is reported too -- the storage + constraint is one reading per deployment/parameter/instant, and a duplicate + inside one request would abort the whole transaction on insert rather than + be reported against the row that caused it. + + Returns the index so the caller can point at the offending row. + """ + for index in range(1, len(observation_datetimes)): + if observation_datetimes[index] <= observation_datetimes[index - 1]: + return index + + return None + + +def spans_overlap( + a_start: datetime, a_end: datetime, b_start: datetime, b_end: datetime +) -> bool: + """Whether two closed intervals share any instant. See the module docstring.""" + return not (a_end < b_start or a_start > b_end) + + +def resolve_deployment_id( + candidates: list[tuple[int, date | None, date | None]], + span_start: datetime, + span_end: datetime, +) -> int: + """ + Pick the deployment whose installation period covers a block's span. + + ``candidates`` are ``(deployment_id, installation_date, removal_date)`` for + one well. A NULL installation date reads as "always installed" and a NULL + removal date as "still installed" -- that is how the column is used, and + treating an unrecorded date as a closed boundary would exclude the + deployments most likely to be current. + + Ambiguity is an error rather than a choice: two overlapping deployments mean + two sensors could have produced the file, and guessing attributes readings to + hardware that did not record them. + """ + span_start_date = span_start.date() + span_end_date = span_end.date() + + covering = [ + deployment_id + for deployment_id, installation_date, removal_date in candidates + if (installation_date is None or installation_date <= span_start_date) + and (removal_date is None or removal_date >= span_end_date) + ] + + if not covering: + raise HydrographError( + f"No deployment covers {span_start_date} to {span_end_date}; " + "send deployment_id explicitly" + ) + if len(covering) > 1: + joined = ", ".join(str(deployment_id) for deployment_id in sorted(covering)) + raise HydrographError( + f"{len(covering)} deployments cover {span_start_date} to " + f"{span_end_date} ({joined}); send deployment_id explicitly" + ) + + return covering[0] + + +def narrowed_block_span( + surviving_datetimes: list[datetime], +) -> tuple[datetime, datetime] | None: + """ + What a block's span becomes after some of its readings are deleted. + + ``None`` means nothing survived and the block should go with them -- an + empty block is invisible to the reader and would only ever collide with a + later publish. + """ + if not surviving_datetimes: + return None + + return min(surviving_datetimes), max(surviving_datetimes) + + +def validate_delete_range(start_time: datetime, end_time: datetime) -> None: + """ + Reject a delete range that cannot be meant. + + Both bounds are required by the route signature; this covers the ordering. + An inverted range is not silently reordered, because the operation is + irreversible and a transposed pair is as likely to be the wrong pair as the + right one written backwards. + """ + if end_time <= start_time: + raise HydrographError("end_time must be after start_time") + + +# ============= EOF ============================================= diff --git a/schemas/transducer.py b/schemas/transducer.py index f11be79aa..a9c5c4142 100644 --- a/schemas/transducer.py +++ b/schemas/transducer.py @@ -14,10 +14,12 @@ # limitations under the License. # =============================================================================== from datetime import datetime +from typing import Literal -from pydantic import BaseModel +from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, field_validator -from core.enums import DataMaturity, ReviewStatus +from core.enums import DataMaturity, ReleaseStatus, ReviewStatus +from domain.hydrograph import MAX_MEASUREMENTS, first_out_of_order_index from schemas import BaseResponseModel, BaseCreateModel @@ -28,12 +30,22 @@ class TransducerObservationBlockResponse(BaseResponseModel): parameter_id: int # parameter: ParameterResponse + # Publish provenance. Nullable throughout: blocks loaded from the legacy + # AMPAPI transfer predate the corrector and state none of this. + source_file: str | None = None + source_kind: str | None = None + corrections: list[str] | None = None + comment: str | None = None + class TransducerObservationResponse(BaseResponseModel): value: float observation_datetime: datetime parameter_id: int deployment_id: int + # Set only where a correction moved the value, so NULL reads as + # "as measured" rather than "unknown". + note: str | None = None # Nullable: readings loaded before the field existed do not state a # maturity, and asserting one for them would be an invention. data_maturity: DataMaturity | None @@ -53,4 +65,118 @@ class CreateTransducerObservation(BaseCreateModel): data_maturity: DataMaturity | None = None +# ============= Hydrograph correction publish ==================== +# The corrector (OcotilloUI /ocotillo/hydrograph-correction) uploads one +# corrected logger file as one block. See +# docs/hydrograph-correction-publish.md. + + +class TransducerBlockProvenance(BaseModel): + """Where a corrected series came from and what was done to it.""" + + source_file: str = Field(max_length=255) + source_kind: Literal["water_head", "depth_to_water"] | None = None + # Free text in applied order, written by the workbench: "shift (-1.25 ft, + # ...)", "snap_to_manual (+0.42 ft to ..., collected by ...)". + corrections: list[str] = Field(default_factory=list) + notes: str | None = None + + +class CorrectedMeasurement(BaseModel): + """One reading of the corrected series, in feet below ground surface.""" + + # Aware, so a naive timestamp is rejected rather than guessed at. The + # workbench sends UTC; a logger file's local wall time silently read as UTC + # would shift a whole series by hours. + observation_datetime: AwareDatetime + value: float + note: str | None = None + + # NaN and infinity are not measurements. Without this they would validate + # as floats and land in the column. + model_config = ConfigDict(allow_inf_nan=False) + + +class PublishTransducerBlock(BaseCreateModel): + """A whole corrected file: one block plus every reading in it.""" + + thing_id: int + # Optional: resolved server-side from the block span when omitted, and 422 + # when that is ambiguous. See domain.hydrograph.resolve_deployment_id. + deployment_id: int | None = None + parameter_id: int + + # release_status comes from BaseCreateModel, defaulting to "draft". + review_status: ReviewStatus = "not reviewed" + + provenance: TransducerBlockProvenance + measurements: list[CorrectedMeasurement] = Field( + min_length=1, max_length=MAX_MEASUREMENTS + ) + + @field_validator("review_status", mode="before") + @classmethod + def coerce_review_status(cls, v): + if isinstance(v, str): + try: + return ReviewStatus(v) + except ValueError: + raise ValueError(f"Invalid review_status: {v}") + return v + + @field_validator("measurements") + @classmethod + def measurements_strictly_increasing(cls, measurements): + # Reported against the offending row's index so the UI can highlight + # it: the error path becomes + # ["body", "measurements", N, "observation_datetime"]. + index = first_out_of_order_index( + [measurement.observation_datetime for measurement in measurements] + ) + if index is not None: + raise ValueError( + f"measurements must be strictly increasing in time; row {index} " + f"({measurements[index].observation_datetime.isoformat()}) does not " + f"advance on row {index - 1} " + f"({measurements[index - 1].observation_datetime.isoformat()})" + ) + return measurements + + +class PublishedTransducerBlockResponse(BaseModel): + """ + Mirrors the read shape so the client can merge a publish straight into a + ``GET /observation/transducer-groundwater-level`` result set. The + observations are not echoed -- the client just sent them; the count is what + it cannot know. + """ + + block: TransducerObservationBlockResponse + observation_count: int + thing_id: int + deployment_id: int + + +class OverlappingBlock(BaseModel): + """An existing block a publish would collide with, named in the 409 body.""" + + id: int + start_datetime: datetime + end_datetime: datetime + review_status: ReviewStatus + release_status: ReleaseStatus + + +class DeletedTransducerObservationsResponse(BaseModel): + """ + What a range delete removed. ``updated_block_ids`` are blocks that kept + some readings and had their span narrowed to the survivors. + """ + + deleted_observation_count: int + deleted_block_ids: list[int] + updated_block_ids: list[int] + thing_id: int + + # ============= EOF ============================================= diff --git a/services/observation_helper.py b/services/observation_helper.py index 4e1cab5e6..e01735029 100644 --- a/services/observation_helper.py +++ b/services/observation_helper.py @@ -6,9 +6,9 @@ from fastapi import Request, Query from fastapi_pagination.ext.sqlalchemy import paginate from pydantic import BaseModel -from sqlalchemy import select, desc +from sqlalchemy import asc, select, desc from sqlalchemy.orm import Session -from starlette.status import HTTP_404_NOT_FOUND +from starlette.status import HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_CONTENT from db import ( Observation, @@ -187,11 +187,54 @@ def transformer(observations): return response_items - query = query.order_by(TransducerObservation.observation_datetime.desc()) + query = _sorted_transducer_query(query, sort, order) return paginate(query=query, conn=session, transformer=transformer) +# Only columns that mean something to a client of this endpoint. A whitelist +# rather than getattr on the model: the latter would expose every column, +# including the legacy `nma_*` ones, as a public sort key. +_TRANSDUCER_SORT_COLUMNS = { + "observation_datetime": TransducerObservation.observation_datetime, + "value": TransducerObservation.value, + "id": TransducerObservation.id, +} + + +def _sorted_transducer_query(query, sort: str | None, order: str | None): + """ + Apply `sort`/`order` to a transducer observation query. + + Defaults to newest first, which is what the list view wants and what a + client asking for the latest stored reading relies on. An unrecognised sort + field is rejected rather than ignored -- silently returning a differently + ordered page reads as data changing, not as a bad request. + """ + if sort and sort not in _TRANSDUCER_SORT_COLUMNS: + raise PydanticStyleException( + status_code=HTTP_422_UNPROCESSABLE_CONTENT, + detail=[ + { + "loc": ["query", "sort"], + "msg": ( + f"Cannot sort by '{sort}'. Valid fields: " + f"{', '.join(sorted(_TRANSDUCER_SORT_COLUMNS))}" + ), + "type": "value_error", + "input": sort, + } + ], + ) + + column = _TRANSDUCER_SORT_COLUMNS.get( + sort or "observation_datetime", TransducerObservation.observation_datetime + ) + ascending = (order or "desc").lower() == "asc" + + return query.order_by(asc(column) if ascending else desc(column)) + + def get_observations( request: Request, session: Session, diff --git a/services/transducer_helper.py b/services/transducer_helper.py new file mode 100644 index 000000000..1871fde4f --- /dev/null +++ b/services/transducer_helper.py @@ -0,0 +1,459 @@ +# =============================================================================== +# Copyright 2026 ross +# +# 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. +# =============================================================================== +""" +Publish and range-delete for corrected transducer series. + +Orchestration only: load the well, deployment, and colliding blocks, hand the +decisions to ``domain.hydrograph``, persist the result. See +``docs/hydrograph-correction-publish.md`` for the endpoint contract. +""" + +from datetime import datetime + +from sqlalchemy import delete, insert, select, update +from sqlalchemy.orm import Session +from starlette.status import ( + HTTP_404_NOT_FOUND, + HTTP_409_CONFLICT, + HTTP_422_UNPROCESSABLE_CONTENT, +) + +from db import Parameter, Thing +from db.deployment import Deployment +from db.transducer import TransducerObservation, TransducerObservationBlock +from domain.hydrograph import ( + HydrographError, + derive_block_span, + narrowed_block_span, + resolve_deployment_id, + validate_delete_range, +) +from schemas.transducer import ( + DeletedTransducerObservationsResponse, + OverlappingBlock, + PublishedTransducerBlockResponse, + TransducerObservationBlockResponse, +) +from services.exceptions_helper import PydanticStyleException + +# A published block has not been reviewed by anyone yet, which on USGS terms is +# exactly `provisional`. Once a reviewer marks the block `approved` the readings +# follow. Derived rather than sent by the client so the two axes cannot be set +# to contradict each other on the way in. +_MATURITY_FOR_REVIEW_STATUS = {"approved": "approved"} +_DEFAULT_MATURITY = "provisional" + + +def _not_found(field: str, value, message: str): + return PydanticStyleException( + status_code=HTTP_404_NOT_FOUND, + detail=[ + { + "loc": ["body", field], + "msg": message, + "type": "value_error", + "input": value, + } + ], + ) + + +def _unprocessable(loc: list, value, message: str): + return PydanticStyleException( + status_code=HTTP_422_UNPROCESSABLE_CONTENT, + detail=[ + { + "loc": loc, + "msg": message, + "type": "value_error", + "input": value, + } + ], + ) + + +def _enum_value(value): + """Unwrap a lexicon-backed enum member to the term the column stores.""" + return getattr(value, "value", value) + + +def _deployment_ids_for_thing(session: Session, thing_id: int) -> list[int]: + return list( + session.scalars( + select(Deployment.id).where(Deployment.thing_id == thing_id) + ).all() + ) + + +def _overlapping_blocks( + session: Session, + thing_id: int, + parameter_id: int, + span_start: datetime, + span_end: datetime, +) -> list[TransducerObservationBlock]: + """ + Blocks whose closed span shares an instant with ``[span_start, span_end]``. + + Inclusive on both ends, matching the block reader rather than + ``TransducerObservationBlock.overlaps``. See ``domain.hydrograph``. + """ + return list( + session.scalars( + select(TransducerObservationBlock) + .where( + TransducerObservationBlock.thing_id == thing_id, + TransducerObservationBlock.parameter_id == parameter_id, + TransducerObservationBlock.start_datetime <= span_end, + TransducerObservationBlock.end_datetime >= span_start, + ) + .order_by(TransducerObservationBlock.start_datetime) + ).all() + ) + + +def publish_transducer_block( + session: Session, + payload, + user=None, + replace_overlapping: bool = False, +) -> PublishedTransducerBlockResponse: + """ + Create one block and all of its readings, or nothing. + + The block's span is derived from the readings -- a client-supplied span + wider than the data would attach unrelated readings to the block. + """ + thing = session.get(Thing, payload.thing_id) + if thing is None: + raise _not_found( + "thing_id", payload.thing_id, f"Thing {payload.thing_id} not found" + ) + + parameter = session.get(Parameter, payload.parameter_id) + if parameter is None: + raise _not_found( + "parameter_id", + payload.parameter_id, + f"Parameter {payload.parameter_id} not found", + ) + + span_start, span_end = derive_block_span( + [measurement.observation_datetime for measurement in payload.measurements] + ) + + deployment_id = _resolve_deployment(session, payload, span_start, span_end) + + existing = _overlapping_blocks( + session, payload.thing_id, payload.parameter_id, span_start, span_end + ) + if existing and not replace_overlapping: + raise _overlap_conflict(existing) + + if existing: + _delete_superseded(session, payload.thing_id, payload.parameter_id, existing) + + # Readings can outlive the block that covered them -- nothing links the two + # tables, so a block deleted by hand leaves its observations behind, where + # the reader ignores them but the storage constraint still sees them. Insert + # would abort the transaction on the first collision with a message naming + # the constraint, so check first and say what is actually in the way. + _reject_colliding_observations( + session, deployment_id, payload.parameter_id, span_start, span_end + ) + + block = TransducerObservationBlock( + thing_id=payload.thing_id, + parameter_id=payload.parameter_id, + review_status=_enum_value(payload.review_status), + release_status=_enum_value(payload.release_status), + start_datetime=span_start, + end_datetime=span_end, + source_file=payload.provenance.source_file, + source_kind=payload.provenance.source_kind, + corrections=payload.provenance.corrections or None, + comment=payload.provenance.notes, + ) + _stamp_created_by(block, user) + session.add(block) + session.flush() + + review_status = _enum_value(payload.review_status) + data_maturity = _MATURITY_FOR_REVIEW_STATUS.get(review_status, _DEFAULT_MATURITY) + release_status = _enum_value(payload.release_status) + created_by_id, created_by_name = _created_by(user) + + rows = [ + { + "parameter_id": payload.parameter_id, + "deployment_id": deployment_id, + "observation_datetime": measurement.observation_datetime, + "value": measurement.value, + "note": measurement.note, + "data_maturity": data_maturity, + "release_status": release_status, + "created_by_id": created_by_id, + "created_by_name": created_by_name, + } + for measurement in payload.measurements + ] + session.execute(insert(TransducerObservation), rows) + + session.commit() + session.refresh(block) + + return PublishedTransducerBlockResponse( + block=TransducerObservationBlockResponse.model_validate(block), + observation_count=len(rows), + thing_id=payload.thing_id, + deployment_id=deployment_id, + ) + + +def _resolve_deployment(session, payload, span_start, span_end) -> int: + if payload.deployment_id is not None: + deployment = session.get(Deployment, payload.deployment_id) + if deployment is None: + raise _not_found( + "deployment_id", + payload.deployment_id, + f"Deployment {payload.deployment_id} not found", + ) + if deployment.thing_id != payload.thing_id: + raise _unprocessable( + ["body", "deployment_id"], + payload.deployment_id, + f"Deployment {payload.deployment_id} belongs to thing " + f"{deployment.thing_id}, not {payload.thing_id}", + ) + return payload.deployment_id + + candidates = session.execute( + select( + Deployment.id, Deployment.installation_date, Deployment.removal_date + ).where(Deployment.thing_id == payload.thing_id) + ).all() + + try: + return resolve_deployment_id( + [tuple(row) for row in candidates], span_start, span_end + ) + except HydrographError as err: + raise _unprocessable(["body", "deployment_id"], None, str(err)) + + +def _overlap_conflict(blocks: list[TransducerObservationBlock]): + overlapping = [ + OverlappingBlock.model_validate(block, from_attributes=True).model_dump( + mode="json" + ) + for block in blocks + ] + ids = ", ".join(str(block.id) for block in blocks) + return PydanticStyleException( + status_code=HTTP_409_CONFLICT, + detail=[ + { + "loc": ["body", "measurements"], + "msg": ( + f"Time span overlaps existing block(s) {ids}. Retry with " + "?replace_overlapping=true to supersede them." + ), + "type": "value_error", + "input": {"overlapping_blocks": overlapping}, + } + ], + ) + + +def _delete_superseded( + session: Session, + thing_id: int, + parameter_id: int, + blocks: list[TransducerObservationBlock], +) -> None: + """ + Drop the blocks a replacing publish supersedes, and their readings. + + The readings go too. Keeping them would leave rows the reader cannot show + (no block covers them) that still occupy the deployment/parameter/instant + the new series is about to claim -- so "replace" that kept them would fail + on the very insert it was asked to make room for. + """ + deployment_ids = _deployment_ids_for_thing(session, thing_id) + if deployment_ids: + for block in blocks: + session.execute( + delete(TransducerObservation).where( + TransducerObservation.deployment_id.in_(deployment_ids), + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= block.start_datetime, + TransducerObservation.observation_datetime <= block.end_datetime, + ) + ) + + session.execute( + delete(TransducerObservationBlock).where( + TransducerObservationBlock.id.in_([block.id for block in blocks]) + ) + ) + session.flush() + + +def _reject_colliding_observations( + session: Session, + deployment_id: int, + parameter_id: int, + span_start: datetime, + span_end: datetime, +) -> None: + collisions = session.scalar( + select(TransducerObservation.observation_datetime) + .where( + TransducerObservation.deployment_id == deployment_id, + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= span_start, + TransducerObservation.observation_datetime <= span_end, + ) + .order_by(TransducerObservation.observation_datetime) + .limit(1) + ) + if collisions is None: + return + + raise PydanticStyleException( + status_code=HTTP_409_CONFLICT, + detail=[ + { + "loc": ["body", "measurements"], + "msg": ( + f"Deployment {deployment_id} already has readings in this " + f"time span (earliest {collisions.isoformat()}) that no block " + "covers. Delete them by range before publishing." + ), + "type": "value_error", + "input": {"deployment_id": deployment_id}, + } + ], + ) + + +def delete_transducer_observations( + session: Session, + thing_id: int, + parameter_id: int, + start_time: datetime, + end_time: datetime, +) -> DeletedTransducerObservationsResponse: + """ + Delete every reading for a well inside a closed time range, then reconcile + the blocks that covered them. + + Scoped exactly like the ``GET`` on the same path, so the set a client + previews is the set this removes. There is deliberately no unbounded form. + """ + thing = session.get(Thing, thing_id) + if thing is None: + raise _not_found("thing_id", thing_id, f"Thing {thing_id} not found") + + try: + validate_delete_range(start_time, end_time) + except HydrographError as err: + raise _unprocessable(["query", "end_time"], end_time.isoformat(), str(err)) + + deployment_ids = _deployment_ids_for_thing(session, thing_id) + if not deployment_ids: + return DeletedTransducerObservationsResponse( + deleted_observation_count=0, + deleted_block_ids=[], + updated_block_ids=[], + thing_id=thing_id, + ) + + # Read the affected blocks before deleting: after the readings are gone + # there is nothing left to identify which blocks covered them. + affected_blocks = _overlapping_blocks( + session, thing_id, parameter_id, start_time, end_time + ) + + deleted = session.execute( + delete(TransducerObservation).where( + TransducerObservation.deployment_id.in_(deployment_ids), + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= start_time, + TransducerObservation.observation_datetime <= end_time, + ) + ) + deleted_observation_count = deleted.rowcount or 0 + session.flush() + + deleted_block_ids: list[int] = [] + updated_block_ids: list[int] = [] + + for block in affected_blocks: + surviving = list( + session.scalars( + select(TransducerObservation.observation_datetime).where( + TransducerObservation.deployment_id.in_(deployment_ids), + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= block.start_datetime, + TransducerObservation.observation_datetime <= block.end_datetime, + ) + ).all() + ) + span = narrowed_block_span(surviving) + + if span is None: + deleted_block_ids.append(block.id) + session.execute( + delete(TransducerObservationBlock).where( + TransducerObservationBlock.id == block.id + ) + ) + continue + + new_start, new_end = span + if (new_start, new_end) != (block.start_datetime, block.end_datetime): + updated_block_ids.append(block.id) + session.execute( + update(TransducerObservationBlock) + .where(TransducerObservationBlock.id == block.id) + .values(start_datetime=new_start, end_datetime=new_end) + ) + + session.commit() + + return DeletedTransducerObservationsResponse( + deleted_observation_count=deleted_observation_count, + deleted_block_ids=deleted_block_ids, + updated_block_ids=updated_block_ids, + thing_id=thing_id, + ) + + +def _created_by(user) -> tuple[str | None, str | None]: + if isinstance(user, dict): + return user.get("sub"), user.get("name") + return None, None + + +def _stamp_created_by(obj, user) -> None: + created_by_id, created_by_name = _created_by(user) + obj.created_by_id = created_by_id + obj.created_by_name = created_by_name + + +# ============= EOF ============================================= diff --git a/tests/test_authorization.py b/tests/test_authorization.py index 97608ae9c..02ce5caee 100644 --- a/tests/test_authorization.py +++ b/tests/test_authorization.py @@ -68,6 +68,7 @@ dependencies.amp_admin_function, dependencies.amp_editor_function, dependencies.amp_viewer_function, + dependencies.amp_staging_function, dependencies.lexicon_admin_function, dependencies.lexicon_editor_function, dependencies.no_permission_function, diff --git a/tests/test_domain_hydrograph.py b/tests/test_domain_hydrograph.py new file mode 100644 index 000000000..70e857e13 --- /dev/null +++ b/tests/test_domain_hydrograph.py @@ -0,0 +1,179 @@ +# =============================================================================== +# Copyright 2026 ross +# +# 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. +# =============================================================================== +"""Rules behind publishing and range-deleting corrected transducer series.""" + +from datetime import date, datetime, timedelta, timezone + +import pytest + +from domain.hydrograph import ( + HydrographError, + derive_block_span, + first_out_of_order_index, + narrowed_block_span, + resolve_deployment_id, + spans_overlap, + validate_delete_range, +) + +T0 = datetime(2025, 1, 15, tzinfo=timezone.utc) + + +def _times(*hours): + return [T0 + timedelta(hours=h) for h in hours] + + +# -------------------------------------------------------------------------- +# derive_block_span +# -------------------------------------------------------------------------- +def test_span_is_the_extent_of_the_readings(): + assert derive_block_span(_times(0, 6, 12)) == (T0, T0 + timedelta(hours=12)) + + +def test_span_does_not_assume_the_readings_arrived_sorted(): + assert derive_block_span(_times(12, 0, 6)) == (T0, T0 + timedelta(hours=12)) + + +def test_a_single_reading_gives_a_zero_width_span(): + # Allowed on purpose: the block check constraint is `end >= start` and the + # reader matches inclusively, so one reading still gets covered. + assert derive_block_span(_times(0)) == (T0, T0) + + +def test_empty_series_is_rejected(): + with pytest.raises(HydrographError): + derive_block_span([]) + + +# -------------------------------------------------------------------------- +# first_out_of_order_index +# -------------------------------------------------------------------------- +def test_increasing_series_is_in_order(): + assert first_out_of_order_index(_times(0, 6, 12)) is None + + +def test_going_backwards_is_reported_at_the_offending_row(): + assert first_out_of_order_index(_times(0, 12, 6)) == 2 + + +def test_a_repeated_timestamp_is_reported_too(): + # Not merely untidy: two readings at one instant collide on the + # deployment/parameter/datetime constraint and would abort the insert. + assert first_out_of_order_index(_times(0, 6, 6)) == 2 + + +def test_a_single_row_cannot_be_out_of_order(): + assert first_out_of_order_index(_times(0)) is None + + +# -------------------------------------------------------------------------- +# spans_overlap +# -------------------------------------------------------------------------- +def test_disjoint_spans_do_not_overlap(): + assert not spans_overlap( + T0, T0 + timedelta(1), T0 + timedelta(2), T0 + timedelta(3) + ) + + +def test_nested_span_overlaps(): + assert spans_overlap( + T0 + timedelta(1), T0 + timedelta(2), T0, T0 + timedelta(days=10) + ) + + +def test_spans_touching_at_one_instant_overlap(): + # The distinguishing case. `TransducerObservationBlock.overlaps` is + # half-open and would call this clear; the reader is inclusive, so both + # blocks would claim a reading at the shared instant. + assert spans_overlap(T0, T0 + timedelta(1), T0 + timedelta(1), T0 + timedelta(2)) + + +# -------------------------------------------------------------------------- +# resolve_deployment_id +# -------------------------------------------------------------------------- +SPAN_START = datetime(2025, 3, 1, tzinfo=timezone.utc) +SPAN_END = datetime(2025, 3, 31, tzinfo=timezone.utc) + + +def test_the_one_covering_deployment_is_chosen(): + candidates = [ + (1, date(2020, 1, 1), date(2021, 1, 1)), + (2, date(2025, 1, 1), None), + ] + assert resolve_deployment_id(candidates, SPAN_START, SPAN_END) == 2 + + +def test_a_null_installation_date_reads_as_always_installed(): + assert resolve_deployment_id([(7, None, None)], SPAN_START, SPAN_END) == 7 + + +def test_a_deployment_removed_mid_span_does_not_cover_it(): + with pytest.raises(HydrographError, match="No deployment covers"): + resolve_deployment_id( + [(1, date(2025, 1, 1), date(2025, 3, 15))], SPAN_START, SPAN_END + ) + + +def test_no_deployments_at_all_is_an_error_naming_the_way_out(): + with pytest.raises(HydrographError, match="send deployment_id explicitly"): + resolve_deployment_id([], SPAN_START, SPAN_END) + + +def test_two_covering_deployments_are_ambiguous_rather_than_a_coin_flip(): + with pytest.raises(HydrographError, match="2 deployments cover"): + resolve_deployment_id( + [(1, date(2024, 1, 1), None), (2, None, None)], SPAN_START, SPAN_END + ) + + +# -------------------------------------------------------------------------- +# narrowed_block_span +# -------------------------------------------------------------------------- +def test_a_block_with_nothing_left_is_marked_for_deletion(): + assert narrowed_block_span([]) is None + + +def test_a_block_narrows_to_what_survived(): + assert narrowed_block_span(_times(6, 12)) == ( + T0 + timedelta(hours=6), + T0 + timedelta(hours=12), + ) + + +def test_a_block_down_to_one_reading_becomes_zero_width(): + assert narrowed_block_span(_times(6)) == ( + T0 + timedelta(hours=6), + T0 + timedelta(hours=6), + ) + + +# -------------------------------------------------------------------------- +# validate_delete_range +# -------------------------------------------------------------------------- +def test_a_forward_range_is_accepted(): + assert validate_delete_range(T0, T0 + timedelta(1)) is None + + +def test_an_inverted_range_is_rejected_not_reordered(): + # The operation is irreversible; a transposed pair is as likely to be the + # wrong pair as the right one written backwards. + with pytest.raises(HydrographError, match="after start_time"): + validate_delete_range(T0 + timedelta(1), T0) + + +def test_a_zero_width_range_is_rejected(): + with pytest.raises(HydrographError): + validate_delete_range(T0, T0) diff --git a/tests/test_transducer_publish.py b/tests/test_transducer_publish.py new file mode 100644 index 000000000..9d20a5708 --- /dev/null +++ b/tests/test_transducer_publish.py @@ -0,0 +1,610 @@ +# =============================================================================== +# Copyright 2026 ross +# +# 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. +# =============================================================================== +""" +The hydrograph corrector's publish and range-delete endpoints. + +Both are gated on `AMP.Staging`, which nobody holds in Authentik yet -- these +tests override that dependency, so they cover the behaviour, not the grant. +""" + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import select + +from core.dependencies import amp_staging_function, amp_viewer_function +from db import Deployment, Sensor, Thing, TransducerObservation +from db.engine import session_ctx +from db.transducer import TransducerObservationBlock +from main import app +from tests import client, get_parameter_id, override_authentication + +PUBLISH_URL = "/observation/transducer-groundwater-level/block" +READ_URL = "/observation/transducer-groundwater-level" + +T0 = datetime(2025, 1, 15, tzinfo=timezone.utc) + + +def _groundwater_level_parameter_id() -> int: + return get_parameter_id("groundwater level", "Field Parameter") + + +@pytest.fixture(scope="module", autouse=True) +def override_authentication_dependency_fixture(): + app.dependency_overrides[amp_staging_function] = override_authentication( + default={"name": "foobar", "sub": "1234567890"} + ) + app.dependency_overrides[amp_viewer_function] = override_authentication() + + yield + + app.dependency_overrides = {} + + +@pytest.fixture() +def published_well(): + """A well with one deployment and nothing stored yet.""" + with session_ctx() as session: + thing = Thing( + name=f"Hydrograph Publish Well {datetime.now().timestamp()}", + first_visit_date="2023-03-03", + thing_type="water well", + release_status="draft", + well_depth=200, + hole_depth=200, + well_casing_diameter=5.0, + well_casing_depth=200.0, + ) + sensor = Sensor( + name=f"Hydrograph Publish Sensor {datetime.now().timestamp()}", + sensor_type="Pressure Transducer", + model="Model X", + serial_no=f"serial-{datetime.now().timestamp()}", + pcn_number=f"pcn-{datetime.now().timestamp()}", + owner_agency="NMBGMR", + sensor_status="In Service", + release_status="draft", + ) + session.add_all([thing, sensor]) + session.flush() + + deployment = Deployment( + sensor_id=sensor.id, + thing_id=thing.id, + installation_date="2020-01-01", + removal_date=None, + recording_interval=6, + recording_interval_units="hour", + ) + session.add(deployment) + session.commit() + + thing_id, deployment_id, sensor_id = thing.id, deployment.id, sensor.id + + yield thing_id, deployment_id + + with session_ctx() as session: + deployment_ids = session.scalars( + select(Deployment.id).where(Deployment.thing_id == thing_id) + ).all() + if deployment_ids: + for observation in session.scalars( + select(TransducerObservation).where( + TransducerObservation.deployment_id.in_(deployment_ids) + ) + ).all(): + session.delete(observation) + for block in session.scalars( + select(TransducerObservationBlock).where( + TransducerObservationBlock.thing_id == thing_id + ) + ).all(): + session.delete(block) + session.flush() + for model, pk in ((Deployment, deployment_id), (Thing, thing_id)): + obj = session.get(model, pk) + if obj is not None: + session.delete(obj) + session.flush() + sensor = session.get(Sensor, sensor_id) + if sensor is not None: + session.delete(sensor) + session.commit() + + +def _payload(thing_id, hours=(0, 6, 12), **overrides): + payload = { + "thing_id": thing_id, + "parameter_id": _groundwater_level_parameter_id(), + "release_status": "provisional", + "review_status": "not reviewed", + "provenance": { + "source_file": "SO-0167_20250115.csv", + "source_kind": "water_head", + "corrections": ["convert_water_head (drift corrected)"], + "notes": "Snapped to 2025-01-15 manual measurement.", + }, + "measurements": [ + { + "observation_datetime": (T0 + timedelta(hours=h)).isoformat(), + "value": 42.5 + index * 0.01, + } + for index, h in enumerate(hours) + ], + } + payload.update(overrides) + return payload + + +# -------------------------------------------------------------------------- +# publish +# -------------------------------------------------------------------------- +def test_publish_creates_one_block_and_all_of_its_readings(published_well): + thing_id, deployment_id = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 201, response.text + body = response.json() + assert body["observation_count"] == 3 + assert body["thing_id"] == thing_id + # Deployment resolved server-side: the payload never named one. + assert body["deployment_id"] == deployment_id + + block = body["block"] + assert block["release_status"] == "provisional" + assert block["review_status"] == "not reviewed" + assert block["source_file"] == "SO-0167_20250115.csv" + assert block["corrections"] == ["convert_water_head (drift corrected)"] + assert block["comment"] == "Snapped to 2025-01-15 manual measurement." + + # Span is derived from the data, not sent by the client. + assert block["start_datetime"] == "2025-01-15T00:00:00Z" + assert block["end_datetime"] == "2025-01-15T12:00:00Z" + + +def test_published_readings_come_back_from_the_read_endpoint(published_well): + thing_id, _ = published_well + client.post(PUBLISH_URL, json=_payload(thing_id)) + + response = client.get(READ_URL, params={"thing_id": thing_id}) + + assert response.status_code == 200 + items = response.json()["items"] + assert len(items) == 3 + # Newest first by default, so a client can ask for the latest with size 1. + assert items[0]["observation"]["observation_datetime"] == "2025-01-15T12:00:00Z" + # Unreviewed on publish is provisional on USGS terms. + assert items[0]["observation"]["data_maturity"] == "provisional" + + +def test_per_reading_notes_are_persisted(published_well): + thing_id, _ = published_well + payload = _payload(thing_id) + payload["measurements"][1]["note"] = "spurious reflection removed" + + client.post(PUBLISH_URL, json=payload) + + response = client.get(READ_URL, params={"thing_id": thing_id, "order": "asc"}) + observations = [item["observation"] for item in response.json()["items"]] + assert observations[0]["note"] is None + assert observations[1]["note"] == "spurious reflection removed" + + +def test_a_single_reading_publishes_as_a_zero_width_block(published_well): + thing_id, _ = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id, hours=(0,))) + + assert response.status_code == 201, response.text + block = response.json()["block"] + assert block["start_datetime"] == block["end_datetime"] + + +def test_overlapping_publish_is_rejected_and_names_the_blocks(published_well): + thing_id, _ = published_well + first = client.post(PUBLISH_URL, json=_payload(thing_id)) + first_block_id = first.json()["block"]["id"] + + response = client.post(PUBLISH_URL, json=_payload(thing_id, hours=(6, 18))) + + assert response.status_code == 409 + detail = response.json()["detail"][0] + assert str(first_block_id) in detail["msg"] + assert detail["input"]["overlapping_blocks"][0]["id"] == first_block_id + + +def test_replace_overlapping_supersedes_the_old_block_and_its_readings( + published_well, +): + thing_id, _ = published_well + first = client.post(PUBLISH_URL, json=_payload(thing_id)) + first_block_id = first.json()["block"]["id"] + + response = client.post( + PUBLISH_URL, + params={"replace_overlapping": "true"}, + json=_payload(thing_id, hours=(6, 18)), + ) + + assert response.status_code == 201, response.text + assert response.json()["block"]["id"] != first_block_id + + with session_ctx() as session: + assert session.get(TransducerObservationBlock, first_block_id) is None + + # Only the replacing series survives -- the superseded readings went with + # their block rather than being left where no block covers them. + items = client.get(READ_URL, params={"thing_id": thing_id}).json()["items"] + assert len(items) == 2 + + +def test_publish_is_atomic_when_a_reading_collides(published_well): + thing_id, _ = published_well + client.post(PUBLISH_URL, json=_payload(thing_id)) + + # Delete the block but leave its readings, which is what a hand-deleted + # block leaves behind: rows the reader ignores but storage still holds. + with session_ctx() as session: + for block in session.scalars( + select(TransducerObservationBlock).where( + TransducerObservationBlock.thing_id == thing_id + ) + ).all(): + session.delete(block) + session.commit() + + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 409 + assert "no block covers" in response.json()["detail"][0]["msg"] + + with session_ctx() as session: + blocks = session.scalars( + select(TransducerObservationBlock).where( + TransducerObservationBlock.thing_id == thing_id + ) + ).all() + assert blocks == [] + + +def test_unknown_thing_is_a_404(published_well): + response = client.post(PUBLISH_URL, json=_payload(-1)) + assert response.status_code == 404 + + +def test_out_of_order_measurements_point_at_the_offending_row(published_well): + thing_id, _ = published_well + payload = _payload(thing_id, hours=(0, 12, 6)) + + response = client.post(PUBLISH_URL, json=payload) + + assert response.status_code == 422 + assert any( + error["loc"][:3] == ["body", "measurements"] + for error in response.json()["detail"] + ) + + +def test_naive_timestamps_are_rejected(published_well): + thing_id, _ = published_well + payload = _payload(thing_id) + payload["measurements"][0]["observation_datetime"] = "2025-01-15T00:00:00" + + response = client.post(PUBLISH_URL, json=payload) + + assert response.status_code == 422 + + +def test_an_empty_series_is_rejected(published_well): + thing_id, _ = published_well + response = client.post(PUBLISH_URL, json=_payload(thing_id, hours=())) + assert response.status_code == 422 + + +def test_a_deployment_on_another_well_is_rejected(published_well, sensor): + thing_id, _ = published_well + with session_ctx() as session: + other = Thing( + name=f"Hydrograph Other Well {datetime.now().timestamp()}", + first_visit_date="2023-03-03", + thing_type="water well", + release_status="draft", + ) + session.add(other) + session.flush() + other_deployment = Deployment( + sensor_id=sensor.id, thing_id=other.id, installation_date="2020-01-01" + ) + session.add(other_deployment) + session.commit() + other_deployment_id, other_thing_id = other_deployment.id, other.id + + try: + response = client.post( + PUBLISH_URL, + json=_payload(thing_id, deployment_id=other_deployment_id), + ) + assert response.status_code == 422 + assert "belongs to thing" in response.json()["detail"][0]["msg"] + finally: + with session_ctx() as session: + session.delete(session.get(Deployment, other_deployment_id)) + session.flush() + session.delete(session.get(Thing, other_thing_id)) + session.commit() + + +# -------------------------------------------------------------------------- +# range delete +# -------------------------------------------------------------------------- +def test_deleting_the_whole_span_removes_the_block_too(published_well): + thing_id, _ = published_well + block_id = client.post(PUBLISH_URL, json=_payload(thing_id)).json()["block"]["id"] + + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": T0.isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 200, response.text + body = response.json() + assert body["deleted_observation_count"] == 3 + assert body["deleted_block_ids"] == [block_id] + assert body["updated_block_ids"] == [] + assert client.get(READ_URL, params={"thing_id": thing_id}).json()["items"] == [] + + +def test_a_partial_delete_narrows_the_block_to_the_survivors(published_well): + thing_id, _ = published_well + block_id = client.post(PUBLISH_URL, json=_payload(thing_id)).json()["block"]["id"] + + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": (T0 + timedelta(hours=6)).isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 200, response.text + body = response.json() + assert body["deleted_observation_count"] == 2 + assert body["deleted_block_ids"] == [] + assert body["updated_block_ids"] == [block_id] + + with session_ctx() as session: + block = session.get(TransducerObservationBlock, block_id) + assert block.start_datetime == T0 + assert block.end_datetime == T0 + + # The survivor is still readable, which it would not be if the narrowed + # block no longer covered it. + items = client.get(READ_URL, params={"thing_id": thing_id}).json()["items"] + assert len(items) == 1 + + +def test_an_inverted_delete_range_is_rejected(published_well): + thing_id, _ = published_well + + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": (T0 + timedelta(hours=12)).isoformat(), + "end_time": T0.isoformat(), + }, + ) + + assert response.status_code == 422 + + +def test_delete_without_a_bound_is_rejected(published_well): + thing_id, _ = published_well + + response = client.request("DELETE", READ_URL, params={"thing_id": thing_id}) + + assert response.status_code == 422 + + +def test_delete_for_an_unknown_well_is_a_404(): + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": -1, + "start_time": T0.isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 404 + + +# -------------------------------------------------------------------------- +# read ordering +# -------------------------------------------------------------------------- +def test_ascending_order_is_honoured(published_well): + thing_id, _ = published_well + client.post(PUBLISH_URL, json=_payload(thing_id)) + + items = client.get(READ_URL, params={"thing_id": thing_id, "order": "asc"}).json()[ + "items" + ] + + assert items[0]["observation"]["observation_datetime"] == "2025-01-15T00:00:00Z" + + +def test_an_unknown_sort_field_is_rejected_rather_than_ignored(published_well): + thing_id, _ = published_well + + response = client.get(READ_URL, params={"thing_id": thing_id, "sort": "nonsense"}) + + assert response.status_code == 422 + + +# -------------------------------------------------------------------------- +# deployment resolution at the service boundary +# +# domain.hydrograph covers the rule itself; these cover the translation of its +# verdicts into responses, and the path where the client names a deployment +# and the rule never runs. +# -------------------------------------------------------------------------- +def test_an_explicitly_named_deployment_is_used_as_given(published_well): + thing_id, deployment_id = published_well + + response = client.post( + PUBLISH_URL, json=_payload(thing_id, deployment_id=deployment_id) + ) + + assert response.status_code == 201, response.text + assert response.json()["deployment_id"] == deployment_id + + +def test_unknown_deployment_is_a_404(published_well): + thing_id, _ = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id, deployment_id=-1)) + + assert response.status_code == 404 + assert response.json()["detail"][0]["loc"] == ["body", "deployment_id"] + + +def test_unknown_parameter_is_a_404(published_well): + thing_id, _ = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id, parameter_id=-1)) + + assert response.status_code == 404 + assert response.json()["detail"][0]["loc"] == ["body", "parameter_id"] + + +def test_two_covering_deployments_are_a_422_not_a_coin_flip(published_well, sensor): + thing_id, _ = published_well + with session_ctx() as session: + second = Deployment( + sensor_id=sensor.id, + thing_id=thing_id, + installation_date="2019-01-01", + removal_date=None, + ) + session.add(second) + session.commit() + second_id = second.id + + try: + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 422 + detail = response.json()["detail"][0] + assert detail["loc"] == ["body", "deployment_id"] + assert "2 deployments cover" in detail["msg"] + finally: + with session_ctx() as session: + session.delete(session.get(Deployment, second_id)) + session.commit() + + +def test_a_span_no_deployment_covers_is_a_422(published_well): + thing_id, deployment_id = published_well + # Retire the well's only deployment before the series was recorded. + with session_ctx() as session: + session.get(Deployment, deployment_id).removal_date = "2021-01-01" + session.commit() + + try: + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 422 + assert "No deployment covers" in response.json()["detail"][0]["msg"] + finally: + with session_ctx() as session: + session.get(Deployment, deployment_id).removal_date = None + session.commit() + + +def test_deleting_from_a_well_with_no_deployments_removes_nothing(): + with session_ctx() as session: + thing = Thing( + name=f"Hydrograph Bare Well {datetime.now().timestamp()}", + first_visit_date="2023-03-03", + thing_type="water well", + release_status="draft", + ) + session.add(thing) + session.commit() + thing_id = thing.id + + try: + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": T0.isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 200, response.text + assert response.json() == { + "deleted_observation_count": 0, + "deleted_block_ids": [], + "updated_block_ids": [], + "thing_id": thing_id, + } + finally: + with session_ctx() as session: + session.delete(session.get(Thing, thing_id)) + session.commit() + + +def test_audit_stamping_survives_a_non_dict_user(): + # `authenticated()` yields the token claims, but the development bypass + # yields `True`. Reachable in any environment running with + # AUTHENTIK_DISABLE_AUTHENTICATION, where a publish must still write rather + # than fail reaching into a bool for a subject id. + from services.transducer_helper import _created_by + + assert _created_by(True) == (None, None) + assert _created_by({"sub": "1234567890", "name": "foobar"}) == ( + "1234567890", + "foobar", + ) + + +def test_a_review_status_outside_the_lexicon_is_rejected(published_well): + # `review_status` is lexicon-backed, so an unknown term would otherwise + # reach the column and fail on a foreign key rather than as a bad request. + thing_id, _ = published_well + + response = client.post( + PUBLISH_URL, json=_payload(thing_id, review_status="mostly reviewed") + ) + + assert response.status_code == 422 + assert any( + error["loc"][:2] == ["body", "review_status"] + for error in response.json()["detail"] + ) From ba73c6b012c9c780cb3dc5c3d0bd67e559882867 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 15:49:21 -0700 Subject: [PATCH 2/3] fix(transducer): spell the block time-order constraint correctly The check was created as `check_transuder_block_time_order` -- no `c` -- in the initial migration, so that is the name in every deployed database. Postgres cannot alter a check in place, so the drop-and-recreate that relaxing it to `end_datetime >= start_datetime` already required is the free moment to fix the spelling: dropped under the old name, created under the new one, no separate RENAME and no window where the table is unconstrained beyond the one the relaxation already opens. The two spellings are separate constants because they are not interchangeable. `op.drop_constraint` matches on the name in the live database, so anything reaching for the constraint to *find* it has to use the spelling that matches the revision it is running against -- the old one on the way down, the new one on the way up. Verified by round-tripping the migration against the test database: downgrade restores `check_transuder_...` with `>`, upgrade restores `check_transducer_...` with `>=`. Co-Authored-By: Claude Opus 5 --- ...4e5f6a7b8_hydrograph_correction_publish.py | 24 +++++++++++++++---- db/transducer.py | 2 +- docs/hydrograph-correction-publish.md | 7 ++++++ 3 files changed, 27 insertions(+), 6 deletions(-) diff --git a/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py b/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py index d3f20e363..48e6bebdf 100644 --- a/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py +++ b/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py @@ -24,6 +24,12 @@ matches observations inclusively on both bounds, so a zero-width block still covers its reading. Loosening a check constraint cannot invalidate existing rows. + +That check also gets its name spelled right on the way through. It was created +as `check_transuder_block_time_order` -- no `c` -- and since Postgres cannot +alter a check in place, the drop-and-recreate this migration already performs +is the free moment to fix it. The old name is dropped and the new one created; +no separate RENAME is needed. """ import sqlalchemy as sa @@ -35,9 +41,14 @@ branch_labels = None depends_on = None -# Spelled as it exists in the database, typo included -- renaming it here would -# leave deployed environments with a constraint this migration cannot find. -TIME_ORDER_CONSTRAINT = "check_transuder_block_time_order" +# The name as created by the initial migration, misspelled. Drops and +# downgrades have to use it verbatim: `op.drop_constraint` matches on the name +# in the live database, so correcting the spelling anywhere it is used to +# *find* the constraint would make the statement a no-op target and fail. +LEGACY_TIME_ORDER_CONSTRAINT = "check_transuder_block_time_order" + +# What it is called from this migration forward. +TIME_ORDER_CONSTRAINT = "check_transducer_block_time_order" def upgrade() -> None: @@ -81,8 +92,11 @@ def upgrade() -> None: ), ) + # Dropped under the old name, recreated under the new one: the rename and + # the relaxation are the same statement pair, so there is no window where + # the table is unconstrained beyond the one this already needs. op.drop_constraint( - TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check" + LEGACY_TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check" ) op.create_check_constraint( TIME_ORDER_CONSTRAINT, @@ -105,7 +119,7 @@ def downgrade() -> None: TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check" ) op.create_check_constraint( - TIME_ORDER_CONSTRAINT, + LEGACY_TIME_ORDER_CONSTRAINT, "transducer_observation_block", "end_datetime > start_datetime", ) diff --git a/db/transducer.py b/db/transducer.py index 65766f6e5..e129cfe60 100644 --- a/db/transducer.py +++ b/db/transducer.py @@ -114,7 +114,7 @@ class TransducerObservationBlock(Base, AutoBaseMixin, ReleaseMixin): # observations inclusively on both bounds, so a zero-width block still # covers its reading. CheckConstraint( - "end_datetime >= start_datetime", name="check_transuder_block_time_order" + "end_datetime >= start_datetime", name="check_transducer_block_time_order" ), Index( "ix_transducer_block_time", diff --git a/docs/hydrograph-correction-publish.md b/docs/hydrograph-correction-publish.md index 572e4485f..768b91559 100644 --- a/docs/hydrograph-correction-publish.md +++ b/docs/hydrograph-correction-publish.md @@ -80,6 +80,13 @@ reading becomes zero-width, which the `end_datetime >= start_datetime` check constraint allows on purpose (migration `c3d4e5f6a7b8`) and which the inclusive reader still covers. +That same migration renames the constraint from `check_transuder_...` to +`check_transducer_...`. Postgres cannot alter a check in place, so the +drop-and-recreate the relaxation already required was the free moment to fix +the spelling. The downgrade puts the old name back, so anything reaching for +the constraint by name has to pick the spelling that matches the revision it is +running against. + **This leaves the `transducer_daily_data` materialized view stale** until its next scheduled refresh. Nothing here refreshes it — a full refresh on every delete would cost far more than the correctness it buys between nightly runs. From 4701979f8b4240e1bd1db53e820626476344e8df Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 22:48:27 -0700 Subject: [PATCH 3/3] fix(transducer): serialize series writes and scope the publish parameter Four review findings from the Copilot pass, all real. The publish and range-delete paths each read state, decide, and then write based on what they read, with nothing stopping another writer in between. Two publishes with different timestamps but overlapping spans each saw no existing block and both committed -- the unique constraints only catch identical spans and identical readings -- leaving the inclusive reader with two blocks claiming the same instants. Two range deletes each computed survivors from a snapshot the other was invalidating, so the later update could widen a block back over readings the earlier one had removed. Both now take a transaction-scoped advisory lock on (thing_id, parameter_id) before reading. An advisory lock rather than row locks because on publish there is no row to lock: the conflict is with a block that does not exist yet, so what needs guarding is the series itself. Both paths take the same key in the same order, so they serialize against each other and cannot deadlock against one another. Covered by a test that holds the lock from a second connection and asserts the publish waits for it rather than proceeding on a stale snapshot. `parameter_id` was accepted from the client unchecked while the read and delete routes on the same path resolve groundwater level themselves, so publishing under any other parameter returned 201 for data neither of them could ever list or remove. The field stays -- the contract has the client state it explicitly rather than inherit a server-side default -- but it is now validated against the parameter the route is scoped to. `order` accepted anything and silently fell through to descending, so the near miss `order=ascending` returned 200 with the rows in exactly the opposite order to the one requested. It is now a 422 in the same shape as an unknown sort field. Co-Authored-By: Claude Opus 5 --- api/observation.py | 6 ++- docs/hydrograph-correction-publish.md | 31 +++++++++++- services/observation_helper.py | 25 ++++++++- services/transducer_helper.py | 58 ++++++++++++++++++--- tests/test_transducer_publish.py | 73 +++++++++++++++++++++++++-- 5 files changed, 177 insertions(+), 16 deletions(-) diff --git a/api/observation.py b/api/observation.py index 4e084036c..fb11ca9d7 100644 --- a/api/observation.py +++ b/api/observation.py @@ -137,7 +137,11 @@ def publish_transducer_groundwater_level_block( `docs/hydrograph-correction-publish.md`. """ return publish_transducer_block( - session, payload, user=user, replace_overlapping=replace_overlapping + session, + payload, + parameter_id=_groundwater_level_parameter_id(session), + user=user, + replace_overlapping=replace_overlapping, ) diff --git a/docs/hydrograph-correction-publish.md b/docs/hydrograph-correction-publish.md index 768b91559..3d8a59fbb 100644 --- a/docs/hydrograph-correction-publish.md +++ b/docs/hydrograph-correction-publish.md @@ -46,6 +46,30 @@ transaction. - **Per-reading `note`** is set only where a correction moved the value, so NULL means "as measured" rather than "unknown". +- **`parameter_id` is validated, not obeyed.** The client states it explicitly, + per the contract, but the route checks it against the parameter the route is + scoped to. The read and delete routes on this path resolve groundwater level + themselves, so a block accepted under any other parameter would be a 201 for + data neither of them could ever list or remove. + +### Concurrency + +Both write paths read state, decide, and then write based on what they read, so +each takes a transaction-scoped advisory lock on `(thing_id, parameter_id)` +first — `pg_advisory_xact_lock`. + +Without it, two publishes with different timestamps but overlapping spans each +see no existing block and both commit: the unique constraints only catch +identical spans and identical readings, and the inclusive reader then has two +blocks claiming the same instants. Two range deletes each compute survivors from +a snapshot the other is invalidating, and the later update can widen a block back +over readings the earlier one removed. + +An advisory lock rather than row locks because on publish there is no row to +lock — the conflict is with a block that does not exist yet — so what needs +guarding is the series, not a row. Both paths take the same key, so they +serialize against each other and cannot deadlock against one another. + ### Overlap An existing block for the same well and parameter whose span shares any instant @@ -101,8 +125,11 @@ delete would cost far more than the correctness it buys between nightly runs. keyword-only now. - The read route honours `sort` (`observation_datetime`, `value`, `id`) and `order` (`asc`/`desc`), defaulting to newest first. An unrecognised sort field - is a 422 rather than being ignored — silently returning a differently ordered - page reads as the data changing, not as a bad request. + or order is a 422 rather than being ignored — silently returning a differently + ordered page reads as the data changing, not as a bad request. `order` matters + particularly here: anything other than `asc` used to fall through to + descending, so the near-miss `order=ascending` returned 200 with the rows in + exactly the opposite order to the one asked for. ## Not built diff --git a/services/observation_helper.py b/services/observation_helper.py index e01735029..1afe11158 100644 --- a/services/observation_helper.py +++ b/services/observation_helper.py @@ -201,6 +201,8 @@ def transformer(observations): "id": TransducerObservation.id, } +_TRANSDUCER_SORT_ORDERS = {"asc", "desc"} + def _sorted_transducer_query(query, sort: str | None, order: str | None): """ @@ -227,12 +229,31 @@ def _sorted_transducer_query(query, sort: str | None, order: str | None): ], ) + normalized_order = (order or "desc").lower() + if normalized_order not in _TRANSDUCER_SORT_ORDERS: + # Rejected for the same reason an unknown sort field is: anything other + # than `asc` used to fall through to descending, so `order=ascending` + # returned 200 with the opposite of what was asked for. + raise PydanticStyleException( + status_code=HTTP_422_UNPROCESSABLE_CONTENT, + detail=[ + { + "loc": ["query", "order"], + "msg": ( + f"Cannot order by '{order}'. Valid values: " + f"{', '.join(sorted(_TRANSDUCER_SORT_ORDERS))}" + ), + "type": "value_error", + "input": order, + } + ], + ) + column = _TRANSDUCER_SORT_COLUMNS.get( sort or "observation_datetime", TransducerObservation.observation_datetime ) - ascending = (order or "desc").lower() == "asc" - return query.order_by(asc(column) if ascending else desc(column)) + return query.order_by(asc(column) if normalized_order == "asc" else desc(column)) def get_observations( diff --git a/services/transducer_helper.py b/services/transducer_helper.py index 1871fde4f..99a2e6b5d 100644 --- a/services/transducer_helper.py +++ b/services/transducer_helper.py @@ -23,7 +23,7 @@ from datetime import datetime -from sqlalchemy import delete, insert, select, update +from sqlalchemy import delete, insert, select, text, update from sqlalchemy.orm import Session from starlette.status import ( HTTP_404_NOT_FOUND, @@ -31,7 +31,7 @@ HTTP_422_UNPROCESSABLE_CONTENT, ) -from db import Parameter, Thing +from db import Thing from db.deployment import Deployment from db.transducer import TransducerObservation, TransducerObservationBlock from domain.hydrograph import ( @@ -90,6 +90,32 @@ def _enum_value(value): return getattr(value, "value", value) +def _lock_series(session: Session, thing_id: int, parameter_id: int) -> None: + """ + Serialize writers against one well's series for the rest of the transaction. + + Both the publish overlap check and the delete reconciliation read state, + decide, and then write -- and neither decision survives a concurrent writer. + Two publishes with different timestamps but overlapping spans each see no + existing block and both commit, because the unique constraints only catch + identical spans and identical readings; the inclusive reader then has two + blocks claiming the same instants. Two deletes each compute survivors from + a snapshot the other is invalidating, and the later update can widen a block + back over readings the earlier one removed. + + An advisory lock rather than row locks: on publish there is no row to lock + yet -- the conflict is with a block that does not exist -- so the thing being + guarded is the (well, parameter) series itself, not any row. Transaction + scoped, so it releases on commit or rollback with no unlock path to forget. + Publish and delete take the same key in the same order, so they serialize + against each other and cannot deadlock against one another. + """ + session.execute( + text("SELECT pg_advisory_xact_lock(:thing_id, :parameter_id)"), + {"thing_id": thing_id, "parameter_id": parameter_id}, + ) + + def _deployment_ids_for_thing(session: Session, thing_id: int) -> list[int]: return list( session.scalars( @@ -128,6 +154,7 @@ def _overlapping_blocks( def publish_transducer_block( session: Session, payload, + parameter_id: int, user=None, replace_overlapping: bool = False, ) -> PublishedTransducerBlockResponse: @@ -136,6 +163,10 @@ def publish_transducer_block( The block's span is derived from the readings -- a client-supplied span wider than the data would attach unrelated readings to the block. + + ``parameter_id`` is the parameter this route is scoped to, resolved by the + caller; the payload's own ``parameter_id`` is checked against it rather than + trusted. """ thing = session.get(Thing, payload.thing_id) if thing is None: @@ -143,12 +174,19 @@ def publish_transducer_block( "thing_id", payload.thing_id, f"Thing {payload.thing_id} not found" ) - parameter = session.get(Parameter, payload.parameter_id) - if parameter is None: - raise _not_found( - "parameter_id", + # The read and delete routes on this path resolve the groundwater level + # parameter themselves, so a block published under any other parameter would + # be invisible to both -- a 201 for data that can then never be listed or + # removed here. The field stays in the request because the contract has the + # client state it explicitly rather than inherit a server-side default; it + # is validated, not obeyed. + if payload.parameter_id != parameter_id: + raise _unprocessable( + ["body", "parameter_id"], payload.parameter_id, - f"Parameter {payload.parameter_id} not found", + f"This route publishes parameter {parameter_id} only; the read and " + f"delete routes on this path would not see parameter " + f"{payload.parameter_id}", ) span_start, span_end = derive_block_span( @@ -157,6 +195,10 @@ def publish_transducer_block( deployment_id = _resolve_deployment(session, payload, span_start, span_end) + # Everything from here reads state and then writes based on it, so no other + # writer may touch this series until the transaction ends. + _lock_series(session, payload.thing_id, payload.parameter_id) + existing = _overlapping_blocks( session, payload.thing_id, payload.parameter_id, span_start, span_end ) @@ -374,6 +416,8 @@ def delete_transducer_observations( except HydrographError as err: raise _unprocessable(["query", "end_time"], end_time.isoformat(), str(err)) + _lock_series(session, thing_id, parameter_id) + deployment_ids = _deployment_ids_for_thing(session, thing_id) if not deployment_ids: return DeletedTransducerObservationsResponse( diff --git a/tests/test_transducer_publish.py b/tests/test_transducer_publish.py index 9d20a5708..44e325b06 100644 --- a/tests/test_transducer_publish.py +++ b/tests/test_transducer_publish.py @@ -20,14 +20,15 @@ tests override that dependency, so they cover the behaviour, not the grant. """ +import threading from datetime import datetime, timedelta, timezone import pytest -from sqlalchemy import select +from sqlalchemy import select, text from core.dependencies import amp_staging_function, amp_viewer_function from db import Deployment, Sensor, Thing, TransducerObservation -from db.engine import session_ctx +from db.engine import engine, session_ctx from db.transducer import TransducerObservationBlock from main import app from tests import client, get_parameter_id, override_authentication @@ -464,6 +465,17 @@ def test_an_unknown_sort_field_is_rejected_rather_than_ignored(published_well): assert response.status_code == 422 +def test_an_unknown_order_is_rejected_rather_than_silently_descending(published_well): + # `ascending` is the obvious near-miss for `asc`, and it used to return 200 + # with the rows in exactly the opposite order to the one asked for. + thing_id, _ = published_well + + response = client.get(READ_URL, params={"thing_id": thing_id, "order": "ascending"}) + + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["query", "order"] + + # -------------------------------------------------------------------------- # deployment resolution at the service boundary # @@ -491,12 +503,28 @@ def test_unknown_deployment_is_a_404(published_well): assert response.json()["detail"][0]["loc"] == ["body", "deployment_id"] -def test_unknown_parameter_is_a_404(published_well): +def test_a_parameter_this_route_does_not_serve_is_rejected(published_well): + # A block published under another parameter would be invisible to the read + # and delete routes on this path, which resolve groundwater level + # themselves -- a 201 for data that could then never be listed or removed + # here. + thing_id, _ = published_well + other_parameter_id = get_parameter_id("pH", "Field Parameter") + + response = client.post( + PUBLISH_URL, json=_payload(thing_id, parameter_id=other_parameter_id) + ) + + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "parameter_id"] + + +def test_an_unknown_parameter_is_rejected_too(published_well): thing_id, _ = published_well response = client.post(PUBLISH_URL, json=_payload(thing_id, parameter_id=-1)) - assert response.status_code == 404 + assert response.status_code == 422 assert response.json()["detail"][0]["loc"] == ["body", "parameter_id"] @@ -608,3 +636,40 @@ def test_a_review_status_outside_the_lexicon_is_rejected(published_well): error["loc"][:2] == ["body", "review_status"] for error in response.json()["detail"] ) + + +def test_publish_waits_for_a_concurrent_writer_on_the_same_series(published_well): + """ + The overlap check reads state and then writes based on it, so it is only + correct if no other writer can slip between the two. Hold the series lock + from another connection and the publish must block rather than proceed on a + snapshot that is already stale. + """ + thing_id, _ = published_well + parameter_id = _groundwater_level_parameter_id() + result = {} + + def publish(): + result["response"] = client.post(PUBLISH_URL, json=_payload(thing_id)) + + blocker = engine.connect() + try: + blocker.execute( + text("SELECT pg_advisory_xact_lock(:thing_id, :parameter_id)"), + {"thing_id": thing_id, "parameter_id": parameter_id}, + ) + + worker = threading.Thread(target=publish, daemon=True) + worker.start() + worker.join(timeout=3) + assert worker.is_alive(), "publish did not wait for the series lock" + + # Releasing the transaction releases the lock. + blocker.rollback() + + worker.join(timeout=30) + assert not worker.is_alive(), "publish never completed after the lock lifted" + finally: + blocker.close() + + assert result["response"].status_code == 201, result["response"].text