From 0afbeee66c05d7bb7ceff3bdaf1bfe97be9a17b6 Mon Sep 17 00:00:00 2001 From: Alex Axthelm Date: Tue, 22 Sep 2026 13:56:41 +0200 Subject: [PATCH 1/6] feat(api): label DB query log events with query_name MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two list-handler queries both open with `WITH resource_universe AS …`, so they were indistinguishable in Azure logs by SQL text alone. Add an optional `query_name` label to `stitch.api.observability.query` events so analysis can identify a query by a stable name instead of SQL-string matching. A `query_name_var` contextvar plus a `named_query()` context manager (observability/context.py) let a call site label the queries it runs; the timing listener reads the var and adds `query_name` to the event only when set, so unlabeled queries omit the field and existing log consumers are unaffected. Because the label rides a contextvar, every statement in the scope — including ORM secondary queries and shared helpers — inherits the name, so only the action-function boundaries are wrapped. Label the resource/source/merge-candidate action entrypoints; the two list queries are distinguishable as resources.list_ids and resources.count, and filter options as filter_options. (e.g. filter_options.country). No behavior change. Docs (deployments/PERFORMANCE.md) and query-timing tests updated. Co-Authored-By: Claude Opus 4.8 --- deployments/PERFORMANCE.md | 10 +- .../stitch/api/db/merge_candidate_actions.py | 82 ++++---- .../api/db/og_field_resource_actions.py | 182 ++++++++++-------- .../stitch/api/db/og_field_source_actions.py | 60 +++--- .../src/stitch/api/observability/context.py | 37 +++- .../stitch/api/observability/query_timing.py | 30 +-- .../tests/observability/test_query_timing.py | 41 +++- 7 files changed, 281 insertions(+), 161 deletions(-) diff --git a/deployments/PERFORMANCE.md b/deployments/PERFORMANCE.md index ab1b0215..f0811c69 100644 --- a/deployments/PERFORMANCE.md +++ b/deployments/PERFORMANCE.md @@ -23,7 +23,7 @@ Two structured log streams, distinguished by the `logger` field: | Logger | Emitted | Key fields | |---|---|---| | `stitch.observability.request` | once per HTTP request (always) | `route`, `method`, `status_code`, `duration_ms`, `db_query_count`, `db_time_ms`, `request_id` | -| `stitch.api.observability.query` | once per query above the slow threshold | `statement` (parameterized SQL, **no bound values**), `duration_ms`, `rowcount`, `route`, `request_id` | +| `stitch.api.observability.query` | once per query above the slow threshold | `statement` (parameterized SQL, **no bound values**), `duration_ms`, `rowcount`, `route`, `request_id`, `query_name` (when the query runs in a labeled scope) | > The request summary is emitted by the shared `stitch.observability` > middleware, so it logs under `stitch.observability.request` (the API's @@ -34,6 +34,14 @@ Two structured log streams, distinguished by the `logger` field: `db_query_count` on a request is the N+1 detector; the `query` stream tells you *which* statement is expensive. +`query_name` is a stable label (e.g. `resources.list_ids`, `resources.count`, +`filter_options.country`) attached to the queries a request handler runs, so you +can pick out a specific query without matching on SQL text — useful when two +statements share a near-identical prefix (both list queries open with +`WITH resource_universe AS …`). It is present only for queries executed inside a +labeled scope; unlabeled queries (e.g. ORM-internal statements outside a handler) +omit the field entirely. + --- ## Step 1 — Enable capture (the knobs) diff --git a/deployments/api/src/stitch/api/db/merge_candidate_actions.py b/deployments/api/src/stitch/api/db/merge_candidate_actions.py index 2162884e..c54af89e 100644 --- a/deployments/api/src/stitch/api/db/merge_candidate_actions.py +++ b/deployments/api/src/stitch/api/db/merge_candidate_actions.py @@ -23,6 +23,7 @@ MergeCandidateStatus, MergeCandidateView, ) +from stitch.api.observability.context import named_query from stitch.ogsi.model import OGFieldSource from stitch.ogsi.model.og_field import OilGasFieldBase from stitch.ogsi.model.types import OGSISrcKey @@ -204,7 +205,8 @@ async def list_merge_candidates(session: AsyncSession) -> list[MergeCandidateVie .options(selectinload(MergeCandidateModel.items)) .order_by(MergeCandidateModel.created.desc()) ) - candidates = (await session.scalars(stmt)).all() + with named_query("merge_candidates.list"): + candidates = (await session.scalars(stmt)).all() return [_candidate_to_view(candidate) for candidate in candidates] @@ -213,7 +215,8 @@ async def get_merge_candidate( candidate_id: int, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> MergeCandidateDetailView: - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.detail"): + candidate = await _load_candidate_model(session, candidate_id) resource_ids = [ item.resource_id for item in sorted(candidate.items, key=lambda i: i.position) @@ -228,14 +231,16 @@ async def get_merge_candidate( # never delete). So a null-shell view always means "emptied by a merge", # never "missing"; no existence check is needed here. Revisit if a resource # hard-delete path is ever added. - by_id = await coalesce_resources_with_sources( - session, resource_ids, licensed_sources - ) + with named_query("merge_candidates.detail"): + by_id = await coalesce_resources_with_sources( + session, resource_ids, licensed_sources + ) # `status` compares the resources' coalesced values; `values` lists every # contributing source tagged with the resource it's attached to, ranked by # the default source order (winner-first). - default_priority = await _default_source_priority(session) + with named_query("merge_candidates.detail"): + default_priority = await _default_source_priority(session) fallback_priority = max(default_priority.values(), default=0) + 1 sources_with_priority = [ (rid, source, default_priority.get(source.source, fallback_priority)) @@ -254,14 +259,16 @@ async def create_merge_candidate( request: MergeCandidateCreateRequest, ) -> MergeCandidateView: resource_ids = _normalize_resource_ids(request.resource_ids) - await _load_mergeable_resources(session, resource_ids) + with named_query("merge_candidates.create"): + await _load_mergeable_resources(session, resource_ids) fingerprint = _fingerprint(resource_ids) - existing = await session.scalar( - select(MergeCandidateModel) - .options(selectinload(MergeCandidateModel.items)) - .where(MergeCandidateModel.fingerprint == fingerprint) - ) + with named_query("merge_candidates.create"): + existing = await session.scalar( + select(MergeCandidateModel) + .options(selectinload(MergeCandidateModel.items)) + .where(MergeCandidateModel.fingerprint == fingerprint) + ) if existing is not None: if existing.status == MergeCandidateStatus.PENDING: raise InvalidActionError( @@ -275,22 +282,23 @@ async def create_merge_candidate( f"An approved merge candidate already exists for resources {resource_ids}." ) - candidate = MergeCandidateModel.create(created_by=user, fingerprint=fingerprint) - session.add(candidate) - await session.flush() + with named_query("merge_candidates.create"): + candidate = MergeCandidateModel.create(created_by=user, fingerprint=fingerprint) + session.add(candidate) + await session.flush() - session.add_all( - [ - MergeCandidateItemModel( - merge_candidate_id=candidate.id, - resource_id=resource_id, - position=position, - ) - for position, resource_id in enumerate(resource_ids) - ] - ) - await session.flush() - await session.refresh(candidate, ["items"]) + session.add_all( + [ + MergeCandidateItemModel( + merge_candidate_id=candidate.id, + resource_id=resource_id, + position=position, + ) + for position, resource_id in enumerate(resource_ids) + ] + ) + await session.flush() + await session.refresh(candidate, ["items"]) return _candidate_to_view(candidate) @@ -300,7 +308,8 @@ async def approve_merge_candidate( candidate_id: int, request: MergeCandidateReviewRequest | None = None, ) -> MergeCandidateView: - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.approve"): + candidate = await _load_candidate_model(session, candidate_id) if candidate.status != MergeCandidateStatus.PENDING: raise InvalidActionError( f"Merge candidate {candidate_id} is not pending; current status={candidate.status}." @@ -309,7 +318,8 @@ async def approve_merge_candidate( resource_ids = [ item.resource_id for item in sorted(candidate.items, key=lambda i: i.position) ] - await _load_mergeable_resources(session, resource_ids) + with named_query("merge_candidates.approve"): + await _load_mergeable_resources(session, resource_ids) merged_resource = await apply_resource_merge( session=session, user=user, @@ -322,9 +332,9 @@ async def approve_merge_candidate( candidate.reviewed_by_id = user.id candidate.last_updated_by_id = user.id candidate.merged_resource_id = merged_resource.id - await session.flush() - - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.approve"): + await session.flush() + candidate = await _load_candidate_model(session, candidate_id) return _candidate_to_view(candidate) @@ -334,7 +344,8 @@ async def deny_merge_candidate( candidate_id: int, request: MergeCandidateReviewRequest | None = None, ) -> MergeCandidateView: - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.deny"): + candidate = await _load_candidate_model(session, candidate_id) if candidate.status != MergeCandidateStatus.PENDING: raise InvalidActionError( f"Merge candidate {candidate_id} is not pending; current status={candidate.status}." @@ -345,6 +356,7 @@ async def deny_merge_candidate( candidate.reviewed_at = datetime.now(timezone.utc) candidate.reviewed_by_id = user.id candidate.last_updated_by_id = user.id - await session.flush() - candidate = await _load_candidate_model(session, candidate_id) + with named_query("merge_candidates.deny"): + await session.flush() + candidate = await _load_candidate_model(session, candidate_id) return _candidate_to_view(candidate) diff --git a/deployments/api/src/stitch/api/db/og_field_resource_actions.py b/deployments/api/src/stitch/api/db/og_field_resource_actions.py index 540c4a75..1b031e52 100644 --- a/deployments/api/src/stitch/api/db/og_field_resource_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_resource_actions.py @@ -13,6 +13,7 @@ ResourceNotFoundError, ) from stitch.api.auth import CurrentUser +from stitch.api.observability.context import named_query from stitch.api.entities import ( FilterOptionField, OGFieldFilterOptionsParams, @@ -77,16 +78,19 @@ async def query( ids_stmt = base_resource_query(params, licensed_sources) count_stmt = select(func.count()).select_from(ids_stmt.subquery()) - total = (await session.scalar(count_stmt)) or 0 + with named_query("resources.count"): + total = (await session.scalar(count_stmt)) or 0 ids_stmt = ids_stmt.limit(params.limit).offset(params.offset) - ids = list((await session.scalars(ids_stmt)).all()) + with named_query("resources.list_ids"): + ids = list((await session.scalars(ids_stmt)).all()) if not ids: return [], total # Hydrate the page with the shared SQL coalescer (same one the detail path # uses), then the shared list-item projection, in phase-1 order. - coalesced = await coalesce_resources(session, ids, licensed_sources) + with named_query("resources.list_hydrate"): + coalesced = await coalesce_resources(session, ids, licensed_sources) items = [resource_to_list_item_view(coalesced[rid]) for rid in ids] return items, total @@ -114,7 +118,8 @@ async def filter_options( value_col = getattr(ranked.c, value_attr_for(params.field)) labeled = value_col.label("value") stmt = select(labeled).where(value_col.is_not(None)).distinct().order_by(labeled) - values = await session.scalars(stmt) + with named_query(f"filter_options.{params.field}"): + values = await session.scalars(stmt) return list(values.all()) @@ -123,20 +128,22 @@ async def get( id: int, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> OGFieldResource: - stmt = ( - select(ResourceModel) - .options(selectinload(ResourceModel.memberships)) - .where(ResourceModel.id == id) - ) - model = await session.scalar(stmt) - if model is None: - raise HTTPException( - status_code=HTTP_404_NOT_FOUND, detail=f"No Resource with id `{id}` found." + with named_query("resources.detail"): + stmt = ( + select(ResourceModel) + .options(selectinload(ResourceModel.memberships)) + .where(ResourceModel.id == id) + ) + model = await session.scalar(stmt) + if model is None: + raise HTTPException( + status_code=HTTP_404_NOT_FOUND, + detail=f"No Resource with id `{id}` found.", + ) + await session.refresh(model, ["memberships"]) + return await resource_model_to_entity( + session, model, licensed_sources=licensed_sources ) - await session.refresh(model, ["memberships"]) - return await resource_model_to_entity( - session, model, licensed_sources=licensed_sources - ) async def resolve_root_id(session: AsyncSession, id: int) -> int: @@ -152,23 +159,29 @@ async def resolve_root_id(session: AsyncSession, id: int) -> int: (``apply_resource_merge`` always targets a brand-new row), so ``get_root`` terminates. """ - model = await session.scalar(select(ResourceModel).where(ResourceModel.id == id)) - if model is None: - raise HTTPException( - status_code=HTTP_404_NOT_FOUND, detail=f"No Resource with id `{id}` found." + with named_query("resources.resolve_root"): + model = await session.scalar( + select(ResourceModel).where(ResourceModel.id == id) ) - if model.repointed_id is None: - return id - try: - return (await model.get_root(session)).id - except ResourceNotFoundError as exc: - # Unreachable under current invariants (FK + acyclic merges + the - # self-reference validator), but a corrupt repoint chain must not surface - # as an unhandled 500 on a read path. - raise HTTPException( - status_code=HTTP_404_NOT_FOUND, - detail=f"Resource `{id}` could not be resolved to a current resource.", - ) from exc + if model is None: + raise HTTPException( + status_code=HTTP_404_NOT_FOUND, + detail=f"No Resource with id `{id}` found.", + ) + if model.repointed_id is None: + return id + try: + return (await model.get_root(session)).id + except ResourceNotFoundError as exc: + # Unreachable under current invariants (FK + acyclic merges + the + # self-reference validator), but a corrupt repoint chain must not + # surface as an unhandled 500 on a read path. + raise HTTPException( + status_code=HTTP_404_NOT_FOUND, + detail=( + f"Resource `{id}` could not be resolved to a current resource." + ), + ) from exc async def get_resolved( @@ -213,9 +226,12 @@ async def field_source_values( # field_source_candidates ranks in SQL by the same tiered key as the coalesce # winner, so the row order is winner-first and rank is just the enumerate # index. Empty text can't be persisted, so every returned row is a real value. - rows = ( - await session.execute(field_source_candidates(root_id, field, licensed_sources)) - ).all() + with named_query("resources.field_source_values"): + rows = ( + await session.execute( + field_source_candidates(root_id, field, licensed_sources) + ) + ).all() return [ OGFieldSourceValueView( source=row.source, @@ -257,7 +273,8 @@ async def set_field_source_priority( status_code=422, detail=f"field={field} is not a known resource field.", ) - resource = await session.get(ResourceModel, id) + with named_query("resources.set_field_source_priority"): + resource = await session.get(ResourceModel, id) if resource is None: raise HTTPException( status_code=HTTP_404_NOT_FOUND, detail=f"No Resource with id `{id}` found." @@ -282,9 +299,10 @@ async def set_field_source_priority( # Eligibility + current effective order come from the same ranked read the GET # endpoint uses: licensed sources with a non-empty value for the field, # winner-first. - rows = ( - await session.execute(field_source_candidates(id, field, licensed_sources)) - ).all() + with named_query("resources.set_field_source_priority"): + rows = ( + await session.execute(field_source_candidates(id, field, licensed_sources)) + ).all() current_order = [row.source_pk for row in rows] eligible = set(current_order) requested = set(ordered_source_pks) @@ -301,24 +319,25 @@ async def set_field_source_priority( return await field_source_values(session, id, field, licensed_sources) source_by_pk = {row.source_pk: row.source for row in rows} - await session.execute( - delete(OGFieldResourceSourcePriority).where( - OGFieldResourceSourcePriority.resource_id == id, - OGFieldResourceSourcePriority.colname == field, - ) - ) - for priority, source_pk in enumerate(ordered_source_pks): - session.add( - OGFieldResourceSourcePriority.create( - created_by=user, - resource_id=id, - source=source_by_pk[source_pk], - source_pk=source_pk, - colname=field, - priority=priority, + with named_query("resources.set_field_source_priority"): + await session.execute( + delete(OGFieldResourceSourcePriority).where( + OGFieldResourceSourcePriority.resource_id == id, + OGFieldResourceSourcePriority.colname == field, ) ) - await session.flush() + for priority, source_pk in enumerate(ordered_source_pks): + session.add( + OGFieldResourceSourcePriority.create( + created_by=user, + resource_id=id, + source=source_by_pk[source_pk], + source_pk=source_pk, + colname=field, + priority=priority, + ) + ) + await session.flush() return await field_source_values(session, id, field, licensed_sources) @@ -337,17 +356,20 @@ async def create( raise ResourceIntegrityError( f"Cannot create resource that has been repointed.\n\tNew: {repr(resource)}" ) - model = ResourceModel.create(created_by=user) - session.add(model) - await session.flush() - if resource.source_data: - src_models = await get_or_create_sources(session, user, resource.source_data) - res = await attach_sources_to_resource( - session=session, resource_id=model.id, source_rows=src_models, user=user - ) - return res - await session.refresh(model, ["memberships"]) - return await resource_model_to_entity(session, model) + with named_query("resources.create"): + model = ResourceModel.create(created_by=user) + session.add(model) + await session.flush() + if resource.source_data: + src_models = await get_or_create_sources( + session, user, resource.source_data + ) + res = await attach_sources_to_resource( + session=session, resource_id=model.id, source_rows=src_models, user=user + ) + return res + await session.refresh(model, ["memberships"]) + return await resource_model_to_entity(session, model) async def apply_resource_merge( @@ -371,7 +393,8 @@ async def apply_resource_merge( stmt = select(ResourceModel).where(ResourceModel.id.in_(unique_ids)) - results = (await session.scalars(stmt)).all() + with named_query("resources.merge"): + results = (await session.scalars(stmt)).all() missing_ids = set(unique_ids).difference(set([r.id for r in results])) if len(missing_ids) > 0: msg = f"Resources not found for ids: [{','.join(map(str, missing_ids))}]" @@ -392,20 +415,21 @@ async def apply_resource_merge( # originals are intentionally NOT carried over -- merging resets ordering to # default. (No-op reset today since the target is fresh; a later PR handles # an explicit reset if merge semantics ever preserve an existing resource.) - new_resource = ResourceModel.create(created_by=user) - session.add(new_resource) - await session.flush() + with named_query("resources.merge"): + new_resource = ResourceModel.create(created_by=user) + session.add(new_resource) + await session.flush() - # all results are still members of the session - # changes will be picked up on commit - for res in results: - res.repointed_id = new_resource.id + # all results are still members of the session + # changes will be picked up on commit + for res in results: + res.repointed_id = new_resource.id - _ = await _repoint_memberships(session, user, new_resource.id, unique_ids) + _ = await _repoint_memberships(session, user, new_resource.id, unique_ids) - # Return the canonical resource entity - await session.refresh(new_resource, ["memberships"]) - return await resource_model_to_entity(session, new_resource) + # Return the canonical resource entity + await session.refresh(new_resource, ["memberships"]) + return await resource_model_to_entity(session, new_resource) async def _repoint_memberships( diff --git a/deployments/api/src/stitch/api/db/og_field_source_actions.py b/deployments/api/src/stitch/api/db/og_field_source_actions.py index 7924f34c..2abd5eec 100644 --- a/deployments/api/src/stitch/api/db/og_field_source_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_source_actions.py @@ -11,6 +11,7 @@ ) from stitch.api.db.utils import partition_by_id_none from stitch.api.entities import OGFieldQueryParams, User +from stitch.api.observability.context import named_query from stitch.ogsi.model import OGFieldSource, OGFieldResource from stitch.ogsi.model.types import OGSISrcKey @@ -36,7 +37,8 @@ async def create_source( model = OilGasFieldSourceModel.create_from_entity(source, created_by=user) session.add(model) - await session.flush() + with named_query("sources.create"): + await session.flush() return model.as_entity() @@ -95,10 +97,10 @@ async def create_and_attach_sources( ) # Fail fast: validate the target before creating any source, so an invalid # resource_id never leaves a source insert to roll back. - resource = await _get_attachable_resource(session, resource_id) - - models = await _create_source_models(session, user, sources) - await _attach_source_models(session, resource, models, user) + with named_query("sources.create_and_attach"): + resource = await _get_attachable_resource(session, resource_id) + models = await _create_source_models(session, user, sources) + await _attach_source_models(session, resource, models, user) return [model.as_entity() for model in models] @@ -128,10 +130,9 @@ async def get_or_create_sources( data: Sequence[OGFieldSource], ) -> Sequence[OGFieldSource]: - return [ - src.as_entity() - for src in await _get_or_create_source_models(session, user, data) - ] + with named_query("sources.get_or_create"): + models = await _get_or_create_source_models(session, user, data) + return [src.as_entity() for src in models] async def _get_or_create_source_models( @@ -202,15 +203,17 @@ async def attach_sources_to_resource( user: User, ) -> OGFieldResource: """Link an OG field source to a resource via membership.""" - resource = await _get_attachable_resource(session, resource_id) - if len(source_rows) < 1: - raise ResourceIntegrityError( - f"Must pass at least 1 source row to attach to resource (id: `{resource_id}`)." - ) + with named_query("sources.attach"): + resource = await _get_attachable_resource(session, resource_id) + if len(source_rows) < 1: + raise ResourceIntegrityError( + f"Must pass at least 1 source row to attach to resource " + f"(id: `{resource_id}`)." + ) - src_models = await _get_or_create_source_models(session, user, source_rows) - await _attach_source_models(session, resource, src_models, user) - return await resource_model_to_entity(session, resource) + src_models = await _get_or_create_source_models(session, user, source_rows) + await _attach_source_models(session, resource, src_models, user) + return await resource_model_to_entity(session, resource) async def get_source( @@ -218,7 +221,8 @@ async def get_source( id: int, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> OGFieldSource: - model = await session.get(OilGasFieldSourceModel, id) + with named_query("sources.detail"): + model = await session.get(OilGasFieldSourceModel, id) if model is None: raise SourceNotFoundError(f"No OG Field Source found for id `{id}`") if licensed_sources is not None and model.source not in licensed_sources: @@ -239,7 +243,8 @@ async def get_sources( session: AsyncSession, ids: Sequence[int] ) -> Sequence[OGFieldSource]: stmt = select(OilGasFieldSourceModel).where(OilGasFieldSourceModel.id.in_(ids)) - models = (await session.scalars(stmt)).all() + with named_query("sources.get_by_ids"): + models = (await session.scalars(stmt)).all() return [model.as_entity() for model in models] @@ -251,17 +256,20 @@ async def query( """Filtered/sorted/paginated source records (id-ordered) plus total count.""" stmt = base_source_query(params, licensed_sources) count_stmt = select(func.count()).select_from(stmt.subquery()) - total = (await session.scalar(count_stmt)) or 0 + with named_query("sources.count"): + total = (await session.scalar(count_stmt)) or 0 stmt = stmt.limit(params.limit).offset(params.offset) - ids = list((await session.scalars(stmt)).all()) + with named_query("sources.list_ids"): + ids = list((await session.scalars(stmt)).all()) if not ids: return (), total - headers = ( - await session.scalars( - select(OilGasFieldSourceModel).where(OilGasFieldSourceModel.id.in_(ids)) - ) - ).all() + with named_query("sources.list_hydrate"): + headers = ( + await session.scalars( + select(OilGasFieldSourceModel).where(OilGasFieldSourceModel.id.in_(ids)) + ) + ).all() by_id = {h.id: h for h in headers} return tuple(by_id[i].as_entity() for i in ids if i in by_id), total diff --git a/deployments/api/src/stitch/api/observability/context.py b/deployments/api/src/stitch/api/observability/context.py index 4104e70e..b6ea4ad8 100644 --- a/deployments/api/src/stitch/api/observability/context.py +++ b/deployments/api/src/stitch/api/observability/context.py @@ -1,20 +1,25 @@ -"""API-specific per-request DB stats for the query-timing listener. +"""API-specific per-request DB stats and query labels for the timing listener. The generic request context (request id / route / scenario) is owned by the shared ``stitch.observability.middleware``; consumers import those directly from -there. This module owns only the API-specific ``db_stats`` dict: the request -middleware sets it before handing off to the rest of the app, the SQLAlchemy -listener mutates it in place, and the middleware reads back the aggregated query -count / time once the request completes. +there. This module owns the API-specific pieces the query-timing listener reads: +the ``db_stats`` dict (the request middleware sets it before handing off to the +rest of the app, the SQLAlchemy listener mutates it in place, and the middleware +reads back the aggregated query count / time once the request completes) and the +optional ``query_name`` label a call site attaches to the queries it runs. """ +from collections.abc import Iterator +from contextlib import contextmanager from contextvars import ContextVar from typing import TypedDict __all__ = [ "DbStats", "db_stats_var", + "named_query", "new_db_stats", + "query_name_var", ] @@ -25,6 +30,28 @@ class DbStats(TypedDict): db_stats_var: ContextVar[DbStats | None] = ContextVar("stitch_db_stats", default=None) +# Optional label a call site attaches to the queries it runs, so the timing +# listener can tag its events with a stable ``query_name`` instead of relying on +# SQL-text matching. Defaults to ``None`` (unlabeled), and every statement run +# inside a ``named_query`` scope -- including ORM-emitted secondary queries and +# shared helpers -- inherits the active name. +query_name_var: ContextVar[str | None] = ContextVar("stitch_query_name", default=None) + def new_db_stats() -> DbStats: return {"count": 0, "time_ms": 0.0} + + +@contextmanager +def named_query(name: str) -> Iterator[None]: + """Label every DB query run within this scope with ``name``. + + Set at a logical-operation boundary (typically a DB action function). The + label is read by the query-timing listener and added to its emitted events; + it resets on exit so nothing leaks to later queries. + """ + token = query_name_var.set(name) + try: + yield + finally: + query_name_var.reset(token) diff --git a/deployments/api/src/stitch/api/observability/query_timing.py b/deployments/api/src/stitch/api/observability/query_timing.py index ffe53c5c..46eab2f2 100644 --- a/deployments/api/src/stitch/api/observability/query_timing.py +++ b/deployments/api/src/stitch/api/observability/query_timing.py @@ -20,7 +20,7 @@ scenario_var, ) -from .context import db_stats_var +from .context import db_stats_var, query_name_var from .sinks import emit_query_event try: # py3.12+: monotonic, nanosecond resolution @@ -93,16 +93,18 @@ def _after(conn, cursor, statement, parameters, context, executemany): except Exception: # pragma: no cover - driver dependent rowcount = None - emit_query_event( - { - "duration_ms": round(elapsed_ms, 2), - "rowcount": rowcount - if rowcount is not None and rowcount >= 0 - else None, - "executemany": executemany, - "statement": _normalize_statement(statement, statement_max_chars), - "request_id": request_id_var.get(), - "route": route_var.get(), - "scenario": scenario_var.get(), - } - ) + event = { + "duration_ms": round(elapsed_ms, 2), + "rowcount": rowcount if rowcount is not None and rowcount >= 0 else None, + "executemany": executemany, + "statement": _normalize_statement(statement, statement_max_chars), + "request_id": request_id_var.get(), + "route": route_var.get(), + "scenario": scenario_var.get(), + } + # Optional label from the active ``named_query`` scope. Omitted entirely + # when unset so existing log consumers see no new field for unlabeled + # queries. + if (query_name := query_name_var.get()) is not None: + event["query_name"] = query_name + emit_query_event(event) diff --git a/deployments/api/tests/observability/test_query_timing.py b/deployments/api/tests/observability/test_query_timing.py index 1405f97b..6771db79 100644 --- a/deployments/api/tests/observability/test_query_timing.py +++ b/deployments/api/tests/observability/test_query_timing.py @@ -8,7 +8,11 @@ from sqlalchemy.ext.asyncio import create_async_engine from stitch.api.observability import query_timing -from stitch.api.observability.context import db_stats_var, new_db_stats +from stitch.api.observability.context import ( + db_stats_var, + named_query, + new_db_stats, +) from stitch.api.observability.query_timing import _START_KEY from stitch.observability import JsonFormatter, configure_logging @@ -57,6 +61,41 @@ async def test_accumulates_into_request_stats(self, timed_engine): assert stats["count"] == 2 assert stats["time_ms"] >= 0 + @pytest.mark.anyio + async def test_named_query_labels_event(self, timed_engine): + engine, captured = timed_engine + + async with engine.connect() as conn: + with named_query("resources.count"): + await conn.execute(text("SELECT 1")) + + assert len(captured) == 1 + assert captured[0]["query_name"] == "resources.count" + + @pytest.mark.anyio + async def test_query_name_absent_when_unset(self, timed_engine): + engine, captured = timed_engine + + async with engine.connect() as conn: + await conn.execute(text("SELECT 1")) + + assert len(captured) == 1 + # Field is omitted entirely for unlabeled queries so existing log + # consumers see no new key. + assert "query_name" not in captured[0] + + @pytest.mark.anyio + async def test_named_query_scope_resets(self, timed_engine): + engine, captured = timed_engine + + async with engine.connect() as conn: + with named_query("resources.count"): + await conn.execute(text("SELECT 1")) + await conn.execute(text("SELECT 2")) + + assert captured[0]["query_name"] == "resources.count" + assert "query_name" not in captured[1] + @pytest.mark.anyio async def test_respects_slow_query_threshold(self, monkeypatch): captured: list[dict] = [] From 604d909e0bdfbb0cab6be3a1d90c4dfb7074c947 Mon Sep 17 00:00:00 2001 From: Alex Axthelm Date: Tue, 22 Sep 2026 14:36:32 +0200 Subject: [PATCH 2/6] disambiguate multi-query action labels with a third level Several action functions ran multiple distinct statements under one query_name (e.g. merge_candidates.detail wrapped the candidate load, the coalesce, and the default-priority read), so those statements were still indistinguishable by name alone. Give each wrapped statement a third-level suffix (.load / .load_resources / .check_existing / .coalesce / .default_priority / .candidates / .apply / .persist) so every labeled query is uniquely identifiable in logs. No behavior change; single-use labels (resources.count, resources.detail, filter_options., etc.) are unchanged. Co-Authored-By: Claude Opus 4.8 --- .../stitch/api/db/merge_candidate_actions.py | 22 +++++++++---------- .../api/db/og_field_resource_actions.py | 10 ++++----- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/deployments/api/src/stitch/api/db/merge_candidate_actions.py b/deployments/api/src/stitch/api/db/merge_candidate_actions.py index c54af89e..ad9e214e 100644 --- a/deployments/api/src/stitch/api/db/merge_candidate_actions.py +++ b/deployments/api/src/stitch/api/db/merge_candidate_actions.py @@ -215,7 +215,7 @@ async def get_merge_candidate( candidate_id: int, licensed_sources: Collection[OGSISrcKey] | None = None, ) -> MergeCandidateDetailView: - with named_query("merge_candidates.detail"): + with named_query("merge_candidates.detail.load"): candidate = await _load_candidate_model(session, candidate_id) resource_ids = [ @@ -231,7 +231,7 @@ async def get_merge_candidate( # never delete). So a null-shell view always means "emptied by a merge", # never "missing"; no existence check is needed here. Revisit if a resource # hard-delete path is ever added. - with named_query("merge_candidates.detail"): + with named_query("merge_candidates.detail.coalesce"): by_id = await coalesce_resources_with_sources( session, resource_ids, licensed_sources ) @@ -239,7 +239,7 @@ async def get_merge_candidate( # `status` compares the resources' coalesced values; `values` lists every # contributing source tagged with the resource it's attached to, ranked by # the default source order (winner-first). - with named_query("merge_candidates.detail"): + with named_query("merge_candidates.detail.default_priority"): default_priority = await _default_source_priority(session) fallback_priority = max(default_priority.values(), default=0) + 1 sources_with_priority = [ @@ -259,11 +259,11 @@ async def create_merge_candidate( request: MergeCandidateCreateRequest, ) -> MergeCandidateView: resource_ids = _normalize_resource_ids(request.resource_ids) - with named_query("merge_candidates.create"): + with named_query("merge_candidates.create.load_resources"): await _load_mergeable_resources(session, resource_ids) fingerprint = _fingerprint(resource_ids) - with named_query("merge_candidates.create"): + with named_query("merge_candidates.create.check_existing"): existing = await session.scalar( select(MergeCandidateModel) .options(selectinload(MergeCandidateModel.items)) @@ -282,7 +282,7 @@ async def create_merge_candidate( f"An approved merge candidate already exists for resources {resource_ids}." ) - with named_query("merge_candidates.create"): + with named_query("merge_candidates.create.persist"): candidate = MergeCandidateModel.create(created_by=user, fingerprint=fingerprint) session.add(candidate) await session.flush() @@ -308,7 +308,7 @@ async def approve_merge_candidate( candidate_id: int, request: MergeCandidateReviewRequest | None = None, ) -> MergeCandidateView: - with named_query("merge_candidates.approve"): + with named_query("merge_candidates.approve.load"): candidate = await _load_candidate_model(session, candidate_id) if candidate.status != MergeCandidateStatus.PENDING: raise InvalidActionError( @@ -318,7 +318,7 @@ async def approve_merge_candidate( resource_ids = [ item.resource_id for item in sorted(candidate.items, key=lambda i: i.position) ] - with named_query("merge_candidates.approve"): + with named_query("merge_candidates.approve.load_resources"): await _load_mergeable_resources(session, resource_ids) merged_resource = await apply_resource_merge( session=session, @@ -332,7 +332,7 @@ async def approve_merge_candidate( candidate.reviewed_by_id = user.id candidate.last_updated_by_id = user.id candidate.merged_resource_id = merged_resource.id - with named_query("merge_candidates.approve"): + with named_query("merge_candidates.approve.persist"): await session.flush() candidate = await _load_candidate_model(session, candidate_id) return _candidate_to_view(candidate) @@ -344,7 +344,7 @@ async def deny_merge_candidate( candidate_id: int, request: MergeCandidateReviewRequest | None = None, ) -> MergeCandidateView: - with named_query("merge_candidates.deny"): + with named_query("merge_candidates.deny.load"): candidate = await _load_candidate_model(session, candidate_id) if candidate.status != MergeCandidateStatus.PENDING: raise InvalidActionError( @@ -356,7 +356,7 @@ async def deny_merge_candidate( candidate.reviewed_at = datetime.now(timezone.utc) candidate.reviewed_by_id = user.id candidate.last_updated_by_id = user.id - with named_query("merge_candidates.deny"): + with named_query("merge_candidates.deny.persist"): await session.flush() candidate = await _load_candidate_model(session, candidate_id) return _candidate_to_view(candidate) diff --git a/deployments/api/src/stitch/api/db/og_field_resource_actions.py b/deployments/api/src/stitch/api/db/og_field_resource_actions.py index 1b031e52..90139ccb 100644 --- a/deployments/api/src/stitch/api/db/og_field_resource_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_resource_actions.py @@ -273,7 +273,7 @@ async def set_field_source_priority( status_code=422, detail=f"field={field} is not a known resource field.", ) - with named_query("resources.set_field_source_priority"): + with named_query("resources.set_field_source_priority.load"): resource = await session.get(ResourceModel, id) if resource is None: raise HTTPException( @@ -299,7 +299,7 @@ async def set_field_source_priority( # Eligibility + current effective order come from the same ranked read the GET # endpoint uses: licensed sources with a non-empty value for the field, # winner-first. - with named_query("resources.set_field_source_priority"): + with named_query("resources.set_field_source_priority.candidates"): rows = ( await session.execute(field_source_candidates(id, field, licensed_sources)) ).all() @@ -319,7 +319,7 @@ async def set_field_source_priority( return await field_source_values(session, id, field, licensed_sources) source_by_pk = {row.source_pk: row.source for row in rows} - with named_query("resources.set_field_source_priority"): + with named_query("resources.set_field_source_priority.persist"): await session.execute( delete(OGFieldResourceSourcePriority).where( OGFieldResourceSourcePriority.resource_id == id, @@ -393,7 +393,7 @@ async def apply_resource_merge( stmt = select(ResourceModel).where(ResourceModel.id.in_(unique_ids)) - with named_query("resources.merge"): + with named_query("resources.merge.load"): results = (await session.scalars(stmt)).all() missing_ids = set(unique_ids).difference(set([r.id for r in results])) if len(missing_ids) > 0: @@ -415,7 +415,7 @@ async def apply_resource_merge( # originals are intentionally NOT carried over -- merging resets ordering to # default. (No-op reset today since the target is fresh; a later PR handles # an explicit reset if merge semantics ever preserve an existing resource.) - with named_query("resources.merge"): + with named_query("resources.merge.apply"): new_resource = ResourceModel.create(created_by=user) session.add(new_resource) await session.flush() From aa60a443356a2ffee7fbcffb02d260a7c9e15a2a Mon Sep 17 00:00:00 2001 From: Alex Axthelm Date: Tue, 22 Sep 2026 14:51:05 +0200 Subject: [PATCH 3/6] test(api): assert query_name labels at the real action call sites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot review (PR #283): the query-timing unit tests exercised named_query around raw statements but never the wrapped action functions, so a typo'd label or misplaced scope could pass while the acceptance criteria regressed. Add end-to-end tests that register query timing on the integration engine, hit the list / filter-options / detail endpoints, and assert the exact emitted labels — including secondary statements (resources.list_hydrate, resources.resolve_root). Co-Authored-By: Claude Opus 4.8 --- .../observability/test_query_name_actions.py | 95 +++++++++++++++++++ 1 file changed, 95 insertions(+) create mode 100644 deployments/api/tests/observability/test_query_name_actions.py diff --git a/deployments/api/tests/observability/test_query_name_actions.py b/deployments/api/tests/observability/test_query_name_actions.py new file mode 100644 index 00000000..4ade4e22 --- /dev/null +++ b/deployments/api/tests/observability/test_query_name_actions.py @@ -0,0 +1,95 @@ +"""End-to-end coverage that the wrapped DB action call sites emit the expected +``query_name`` labels. + +The unit tests in ``test_query_timing.py`` exercise ``named_query`` around raw +statements; these drive the real endpoints through the router → action → engine +path so a typo in a label (e.g. ``resources.count``) or a misplaced scope is +caught. Query timing is registered on the integration engine with +``log_all_queries=True`` and the sink is captured, so every statement a request +runs is inspected. +""" + +import pytest +from httpx import AsyncClient + +from stitch.api.observability import query_timing + + +@pytest.fixture +def captured_query_events(integration_engine, monkeypatch) -> list[dict]: + """Capture every query event emitted while hitting the integration engine.""" + events: list[dict] = [] + monkeypatch.setattr( + query_timing, "emit_query_event", lambda event: events.append(event) + ) + query_timing.register_query_timing( + integration_engine.sync_engine, slow_query_ms=0, log_all_queries=True + ) + return events + + +def _query_names(events: list[dict]) -> set[str]: + return {event["query_name"] for event in events if "query_name" in event} + + +class TestActionQueryLabels: + @pytest.mark.anyio + async def test_list_endpoint_labels( + self, + integration_client: AsyncClient, + og_create_res_fact, + captured_query_events: list[dict], + ): + # A row must exist for the hydrate query to run (skipped when empty). + create = await integration_client.post( + "/oil-gas-fields/", + json=og_create_res_fact(name="Labeled Resource").model_dump(mode="json"), + ) + assert create.status_code == 200, create.text + + captured_query_events.clear() + response = await integration_client.get("/oil-gas-fields/") + assert response.status_code == 200, response.text + + names = _query_names(captured_query_events) + assert { + "resources.count", + "resources.list_ids", + "resources.list_hydrate", + } <= names + + @pytest.mark.anyio + async def test_filter_options_endpoint_labels( + self, + integration_client: AsyncClient, + captured_query_events: list[dict], + ): + response = await integration_client.get( + "/oil-gas-fields/filter-options", params={"field": "country"} + ) + assert response.status_code == 200, response.text + + assert "filter_options.country" in _query_names(captured_query_events) + + @pytest.mark.anyio + async def test_detail_endpoint_labels( + self, + integration_client: AsyncClient, + og_create_res_fact, + captured_query_events: list[dict], + ): + create = await integration_client.post( + "/oil-gas-fields/", + json=og_create_res_fact(name="Detail Resource").model_dump(mode="json"), + ) + assert create.status_code == 200, create.text + created_id = create.json()["id"] + + captured_query_events.clear() + response = await integration_client.get(f"/oil-gas-fields/{created_id}") + assert response.status_code == 200, response.text + + # get_resolved -> resolve_root_id (resources.resolve_root) then get + # (resources.detail); assert both, so the secondary statement is covered. + names = _query_names(captured_query_events) + assert {"resources.resolve_root", "resources.detail"} <= names From 9dcc1a27abf908a92dbb12c9a81536db10f3f0c4 Mon Sep 17 00:00:00 2001 From: Alex Axthelm Date: Wed, 23 Sep 2026 14:49:39 +0200 Subject: [PATCH 4/6] docs(deployments): note Container App log routing and workspace topology Document where deployed Container App logs go. Add a short "Log routing" note to CI_DEPLOYMENTS.md: routing is set on the Container Apps environment (not the pipeline or app), development + staging lanes feed the non-prod stitch-staging workspace while production is isolated, how to repoint a lane, and why "Parse JSON logs into columns" stays off. Cross-link from PERFORMANCE.md. Co-Authored-By: Claude Opus 4.8 --- deployments/CI_DEPLOYMENTS.md | 20 ++++++++++++++++++++ deployments/PERFORMANCE.md | 3 +++ 2 files changed, 23 insertions(+) diff --git a/deployments/CI_DEPLOYMENTS.md b/deployments/CI_DEPLOYMENTS.md index c5f749ff..497e7ff2 100644 --- a/deployments/CI_DEPLOYMENTS.md +++ b/deployments/CI_DEPLOYMENTS.md @@ -43,11 +43,31 @@ It then handles deployments for: - the database, assuming an existing Azure PostgreSQL flexible server - the API Container App, assuming an existing Container Apps environment + (see "Log routing" below) - the entity-linkage Container App in the same environment - the stitch-llm Container App in the same environment - the ETL Container App (`etl`) in the same environment, on non-`development` lanes only (see below) +### Log routing + +Container App logs (structured JSON on stdout) are forwarded to a Log Analytics +workspace by the **Container Apps environment** (`appLogsConfiguration`), not by +this pipeline or the app — so every app in an environment, including per-PR +preview apps, shares one workspace. Current topology: `development` and +`staging` lanes both feed the non-prod workspace **`stitch-staging`** +(`STITCH-DEV-RG`); `production` is isolated in its own workspace. + +To change where a lane's logs go, edit that lane's environment (named by its +`AZURE_CONTAINER_APP_ENVIRONMENT` variable) — Portal: **Settings → Logging**, or +CLI: `az containerapp env update --logs-destination log-analytics +--logs-workspace-id --logs-workspace-key `. Keep "Parse JSON +logs into columns" **off** on every environment sharing a workspace, so the +table schema stays uniform (the KQL in [`PERFORMANCE.md`](./PERFORMANCE.md) and +`tools/analyze_logs.py` assume the raw-`Log_s` form). This binding lives outside +the repo, so recreating an environment reverts it to `None` (a greyed-out Logs +blade) — reconfigure it when that happens. + ### ETL pipelines (temporary POC wiring) The single `etl` Container App is deployed from a pre-built image published by diff --git a/deployments/PERFORMANCE.md b/deployments/PERFORMANCE.md index f0811c69..1a0eef35 100644 --- a/deployments/PERFORMANCE.md +++ b/deployments/PERFORMANCE.md @@ -7,6 +7,9 @@ the cloud, and readable straight from the terminal locally — so you can find slow/frequent queries from real data instead of guessing. This doc covers the basic loop: **enable capture → drive traffic → analyze**. +For *where* deployed logs are routed (which Log Analytics workspace each lane +feeds, and how to change it), see "Log routing" in +[`CI_DEPLOYMENTS.md`](./CI_DEPLOYMENTS.md). > The instrumentation lives in the app code > ([`deployments/api/src/stitch/api/observability/`](api/src/stitch/api/observability/)), From 73bf7d2913d875dfc82322ea82bb8902ac4178f0 Mon Sep 17 00:00:00 2001 From: Alex Axthelm Date: Thu, 24 Sep 2026 11:27:37 +0200 Subject: [PATCH 5/6] test(api): cover query_name labels at all action call sites; rename filter_options label Address pre-open review findings on PR #283. - Rename the filter-options label filter_options.all -> resources.filter_options so it matches the . scheme every other label uses (the per-field filter_options. from the original AC is obsolete after main's all-fields refactor). Updates the label, PERFORMANCE.md, and the endpoint test. - Add TestActionCallSiteLabels: drive the resource/source/merge-candidate action functions directly and assert the exact query_name for every previously untested call site (write paths, sources.*, merge_candidates.*, and every third-level sub-label), so a typo'd or misplaced named_query fails a test. Co-Authored-By: Claude Opus 4.8 --- deployments/PERFORMANCE.md | 2 +- .../api/db/og_field_resource_actions.py | 2 +- .../observability/test_query_name_actions.py | 258 +++++++++++++++++- 3 files changed, 259 insertions(+), 3 deletions(-) diff --git a/deployments/PERFORMANCE.md b/deployments/PERFORMANCE.md index d8d116c5..5bd2dc21 100644 --- a/deployments/PERFORMANCE.md +++ b/deployments/PERFORMANCE.md @@ -38,7 +38,7 @@ Two structured log streams, distinguished by the `logger` field: *which* statement is expensive. `query_name` is a stable label (e.g. `resources.list_ids`, `resources.count`, -`filter_options.all`) attached to the queries a request handler runs, so you +`resources.filter_options`) attached to the queries a request handler runs, so you can pick out a specific query without matching on SQL text — useful when two statements share a near-identical prefix (both list queries open with `WITH resource_universe AS …`). It is present only for queries executed inside a diff --git a/deployments/api/src/stitch/api/db/og_field_resource_actions.py b/deployments/api/src/stitch/api/db/og_field_resource_actions.py index 46068f05..8485d358 100644 --- a/deployments/api/src/stitch/api/db/og_field_resource_actions.py +++ b/deployments/api/src/stitch/api/db/og_field_resource_actions.py @@ -97,7 +97,7 @@ async def filter_options( ) -> dict[str, list[str]]: """Distinct coalesced values for every filterable field, in one query.""" options: dict[str, list[str]] = {field: [] for field in FILTER_OPTION_FIELDS} - with named_query("filter_options.all"): + with named_query("resources.filter_options"): for colname, value in await session.execute( filter_option_rows(licensed_sources) ): diff --git a/deployments/api/tests/observability/test_query_name_actions.py b/deployments/api/tests/observability/test_query_name_actions.py index cc5f8244..7d225575 100644 --- a/deployments/api/tests/observability/test_query_name_actions.py +++ b/deployments/api/tests/observability/test_query_name_actions.py @@ -11,8 +11,21 @@ import pytest from httpx import AsyncClient +from sqlalchemy.ext.asyncio import AsyncSession +from stitch.api.db import merge_candidate_actions as mca +from stitch.api.db import og_field_resource_actions as resource_actions +from stitch.api.db import og_field_source_actions as source_actions +from stitch.api.db.model import MembershipModel, MembershipStatus, ResourceModel +from stitch.api.db.model.oil_gas_field_source_value import ATTRIBUTE_NAMES +from stitch.api.entities import ( + MergeCandidateCreateRequest, + MergeCandidateReviewRequest, + OGFieldQueryParams, + User, +) from stitch.api.observability import query_timing +from tests.utils import make_source_model @pytest.fixture @@ -68,7 +81,7 @@ async def test_filter_options_endpoint_labels( response = await integration_client.get("/oil-gas-fields/filter-options") assert response.status_code == 200, response.text - assert "filter_options.all" in _query_names(captured_query_events) + assert "resources.filter_options" in _query_names(captured_query_events) @pytest.mark.anyio async def test_detail_endpoint_labels( @@ -92,3 +105,246 @@ async def test_detail_endpoint_labels( # (resources.detail); assert both, so the secondary statement is covered. names = _query_names(captured_query_events) assert {"resources.resolve_root", "resources.detail"} <= names + + +async def _attach_source( + session: AsyncSession, user: User, resource_id: int, source: str, **values +) -> int: + """Attach one active source (carrying ``values``) to a resource; return its pk.""" + model = make_source_model(source=source, created_by_id=user.id, **values) + session.add(model) + await session.flush() + session.add( + MembershipModel.create( + created_by=user, + resource_id=resource_id, + source=model.source, + source_pk=model.id, + status=MembershipStatus.ACTIVE, + ) + ) + await session.flush() + return model.id + + +class TestActionCallSiteLabels: + """Drive the DB action functions directly and assert the exact labels they emit. + + The endpoint tests above cover the read paths; these cover the remaining + call sites (write paths, sources, merge candidates, and every third-level + sub-label) so a typo'd or misplaced ``named_query`` string fails a test + instead of shipping a mislabeled event. + """ + + @pytest.mark.anyio + async def test_resource_action_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session + created = await resource_actions.create( + session, test_user, og_create_res_fact(name="Labelled") + ) + await session.commit() + + captured_query_events.clear() + await resource_actions.query(session, OGFieldQueryParams()) + await resource_actions.filter_options(session) + await resource_actions.get_resolved(session, created.id) + await resource_actions.field_source_values( + session, created.id, next(iter(ATTRIBUTE_NAMES)) + ) + + assert { + "resources.count", + "resources.list_ids", + "resources.list_hydrate", + "resources.filter_options", + "resources.resolve_root", + "resources.detail", + "resources.field_source_values", + } <= _query_names(captured_query_events) + + @pytest.mark.anyio + async def test_create_and_merge_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session + + captured_query_events.clear() + first = await resource_actions.create( + session, test_user, og_create_res_fact(name="Merge A") + ) + second = await resource_actions.create( + session, test_user, og_create_res_fact(name="Merge B") + ) + # create (with source_data) fans out to the source helpers too. + assert { + "resources.create", + "sources.get_or_create", + "sources.attach", + } <= _query_names(captured_query_events) + + captured_query_events.clear() + await resource_actions.apply_resource_merge( + session, test_user, [first.id, second.id] + ) + assert { + "resources.merge.load", + "resources.merge.apply", + } <= _query_names(captured_query_events) + + @pytest.mark.anyio + async def test_set_field_source_priority_labels( + self, + seeded_integration_session: AsyncSession, + integration_session_factory, + test_user: User, + captured_query_events: list[dict], + ): + session = seeded_integration_session + resource = ResourceModel.create(created_by=test_user) + session.add(resource) + await session.flush() + rid = resource.id + await _attach_source(session, test_user, rid, "gem", country="Gemland") + await _attach_source(session, test_user, rid, "rmi", country="Rmiland") + await session.commit() + + # Both sources carry `country`, so both are eligible; reverse the current + # winner-first order to force an actual re-prioritization (the persist path). + current = await resource_actions.field_source_values(session, rid, "country") + reordered = [view.source_id for view in reversed(current)] + assert len(reordered) == 2 + + # Run in a fresh session (one per request in prod) so the `.load` get() + # issues SQL rather than hitting this session's identity map. + captured_query_events.clear() + async with integration_session_factory() as fresh: + await resource_actions.set_field_source_priority( + fresh, test_user, rid, "country", reordered + ) + await fresh.commit() + + assert { + "resources.set_field_source_priority.load", + "resources.set_field_source_priority.candidates", + "resources.set_field_source_priority.persist", + } <= _query_names(captured_query_events) + + @pytest.mark.anyio + async def test_source_action_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + source_maker, + captured_query_events: list[dict], + ): + session = seeded_integration_session + parent = await resource_actions.create( + session, test_user, og_create_res_fact(name="Parent") + ) + await session.commit() + + captured_query_events.clear() + created = await source_actions.create_source( + session, test_user, source_maker(managed=False, source="gem") + ) + await source_actions.create_and_attach_sources( + session, test_user, [source_maker(managed=False, source="rmi")], parent.id + ) + await source_actions.get_or_create_sources( + session, test_user, [source_maker(managed=False, source="wm")] + ) + await source_actions.attach_sources_to_resource( + session, parent.id, [source_maker(managed=False, source="bc")], test_user + ) + await source_actions.get_source(session, created.id) + await source_actions.get_sources(session, [created.id]) + await source_actions.query(session, OGFieldQueryParams()) + + assert { + "sources.create", + "sources.create_and_attach", + "sources.get_or_create", + "sources.attach", + "sources.detail", + "sources.get_by_ids", + "sources.count", + "sources.list_ids", + "sources.list_hydrate", + } <= _query_names(captured_query_events) + + @pytest.mark.anyio + async def test_merge_candidate_action_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session + resources = [ + await resource_actions.create( + session, test_user, og_create_res_fact(name=f"Cand {i}") + ) + for i in range(4) + ] + await session.commit() + + captured_query_events.clear() + approved = await mca.create_merge_candidate( + session, + test_user, + MergeCandidateCreateRequest( + resource_ids=[resources[0].id, resources[1].id] + ), + ) + await mca.list_merge_candidates(session) + await mca.get_merge_candidate(session, approved.id) + await mca.approve_merge_candidate( + session, + test_user, + approved.id, + MergeCandidateReviewRequest(review_notes="ok"), + ) + + denied = await mca.create_merge_candidate( + session, + test_user, + MergeCandidateCreateRequest( + resource_ids=[resources[2].id, resources[3].id] + ), + ) + await mca.deny_merge_candidate( + session, + test_user, + denied.id, + MergeCandidateReviewRequest(review_notes="no"), + ) + + assert { + "merge_candidates.create.load_resources", + "merge_candidates.create.check_existing", + "merge_candidates.create.persist", + "merge_candidates.list", + "merge_candidates.detail.load", + "merge_candidates.detail.coalesce", + "merge_candidates.detail.default_priority", + "merge_candidates.approve.load", + "merge_candidates.approve.load_resources", + "merge_candidates.approve.persist", + "merge_candidates.deny.load", + "merge_candidates.deny.persist", + # approve delegates to apply_resource_merge + "resources.merge.load", + "resources.merge.apply", + } <= _query_names(captured_query_events) From 16974afe264e204ba7d46281e23b686df5c4e27f Mon Sep 17 00:00:00 2001 From: Alex Axthelm Date: Thu, 24 Sep 2026 11:49:02 +0200 Subject: [PATCH 6/6] test(api): assert exact query_name label set per action MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address PR #283 review: subset checks over a labels-only helper let an unexpectedly-labeled or scope-leaked secondary query pass. Capture each operation on its own and assert the exact label set it emits (missing or extra labels now fail), pinning each action's full fan-out. Unlabeled statements (connection/transaction/ORM-internal) stay excluded by design — the field is optional. Co-Authored-By: Claude Opus 4.8 --- .../observability/test_query_name_actions.py | 234 ++++++++++++------ 1 file changed, 156 insertions(+), 78 deletions(-) diff --git a/deployments/api/tests/observability/test_query_name_actions.py b/deployments/api/tests/observability/test_query_name_actions.py index 7d225575..619d2af3 100644 --- a/deployments/api/tests/observability/test_query_name_actions.py +++ b/deployments/api/tests/observability/test_query_name_actions.py @@ -2,11 +2,18 @@ ``query_name`` labels. The unit tests in ``test_query_timing.py`` exercise ``named_query`` around raw -statements; these drive the real endpoints through the router → action → engine -path so a typo in a label (e.g. ``resources.count``) or a misplaced scope is +statements; these drive the real endpoints and action functions through the +router / action / engine path so a typo in a label or a misplaced scope is caught. Query timing is registered on the integration engine with ``log_all_queries=True`` and the sink is captured, so every statement a request runs is inspected. + +Assertions compare the *exact* set of labels an operation emits (not just a +subset), so a secondary statement that carries an unexpected label -- or a scope +that leaks onto the wrong query -- fails the test. Unlabeled statements +(connection setup, transaction control, ORM-internal reads that legitimately run +outside any ``named_query`` scope) are excluded: the label is optional by design, +and asserting their absence would just pin SQLite/driver noise. """ import pytest @@ -41,11 +48,20 @@ def captured_query_events(integration_engine, monkeypatch) -> list[dict]: return events -def _query_names(events: list[dict]) -> set[str]: +def _labels(events: list[dict]) -> set[str]: + """The set of query_name labels present among captured events.""" return {event["query_name"] for event in events if "query_name" in event} -class TestActionQueryLabels: +def _assert_labels(events: list[dict], expected: set[str]) -> None: + """Assert the labeled events are *exactly* ``expected`` (no missing/extra labels).""" + actual = _labels(events) + assert actual == expected, f"unexpected labels: {actual ^ expected}" + + +class TestEndpointQueryLabels: + """Drive the real HTTP read endpoints and assert the exact labels each emits.""" + @pytest.mark.anyio async def test_list_endpoint_labels( self, @@ -64,12 +80,10 @@ async def test_list_endpoint_labels( response = await integration_client.get("/oil-gas-fields/") assert response.status_code == 200, response.text - names = _query_names(captured_query_events) - assert { - "resources.count", - "resources.list_ids", - "resources.list_hydrate", - } <= names + _assert_labels( + captured_query_events, + {"resources.count", "resources.list_ids", "resources.list_hydrate"}, + ) @pytest.mark.anyio async def test_filter_options_endpoint_labels( @@ -77,11 +91,11 @@ async def test_filter_options_endpoint_labels( integration_client: AsyncClient, captured_query_events: list[dict], ): - # filter-options returns every filterable field in one query. + # filter-options returns every filterable field in one combined query. response = await integration_client.get("/oil-gas-fields/filter-options") assert response.status_code == 200, response.text - assert "resources.filter_options" in _query_names(captured_query_events) + _assert_labels(captured_query_events, {"resources.filter_options"}) @pytest.mark.anyio async def test_detail_endpoint_labels( @@ -101,10 +115,10 @@ async def test_detail_endpoint_labels( response = await integration_client.get(f"/oil-gas-fields/{created_id}") assert response.status_code == 200, response.text - # get_resolved -> resolve_root_id (resources.resolve_root) then get - # (resources.detail); assert both, so the secondary statement is covered. - names = _query_names(captured_query_events) - assert {"resources.resolve_root", "resources.detail"} <= names + # get_resolved -> resolve_root_id then get; both statements labeled. + _assert_labels( + captured_query_events, {"resources.resolve_root", "resources.detail"} + ) async def _attach_source( @@ -128,16 +142,16 @@ async def _attach_source( class TestActionCallSiteLabels: - """Drive the DB action functions directly and assert the exact labels they emit. + """Call the DB action functions directly and assert the exact labels each emits. - The endpoint tests above cover the read paths; these cover the remaining - call sites (write paths, sources, merge candidates, and every third-level - sub-label) so a typo'd or misplaced ``named_query`` string fails a test - instead of shipping a mislabeled event. + Each operation is captured on its own (clear -> call -> assert exact set) so + the fan-out of every action -- including nested labeled helpers -- is pinned, + covering the write paths, sources.*, merge_candidates.*, and every + third-level sub-label that the endpoint tests above do not reach. """ @pytest.mark.anyio - async def test_resource_action_labels( + async def test_resource_read_labels( self, seeded_integration_session: AsyncSession, test_user: User, @@ -152,24 +166,32 @@ async def test_resource_action_labels( captured_query_events.clear() await resource_actions.query(session, OGFieldQueryParams()) + _assert_labels( + captured_query_events, + {"resources.count", "resources.list_ids", "resources.list_hydrate"}, + ) + + captured_query_events.clear() await resource_actions.filter_options(session) + _assert_labels(captured_query_events, {"resources.filter_options"}) + + captured_query_events.clear() await resource_actions.get_resolved(session, created.id) + _assert_labels( + captured_query_events, {"resources.resolve_root", "resources.detail"} + ) + + captured_query_events.clear() await resource_actions.field_source_values( session, created.id, next(iter(ATTRIBUTE_NAMES)) ) - - assert { - "resources.count", - "resources.list_ids", - "resources.list_hydrate", - "resources.filter_options", - "resources.resolve_root", - "resources.detail", - "resources.field_source_values", - } <= _query_names(captured_query_events) + _assert_labels( + captured_query_events, + {"resources.resolve_root", "resources.field_source_values"}, + ) @pytest.mark.anyio - async def test_create_and_merge_labels( + async def test_create_label( self, seeded_integration_session: AsyncSession, test_user: User, @@ -179,27 +201,39 @@ async def test_create_and_merge_labels( session = seeded_integration_session captured_query_events.clear() + await resource_actions.create( + session, test_user, og_create_res_fact(name="Created") + ) + # create() with source_data fans out to the source helpers. + _assert_labels( + captured_query_events, + {"resources.create", "sources.get_or_create", "sources.attach"}, + ) + + @pytest.mark.anyio + async def test_merge_labels( + self, + seeded_integration_session: AsyncSession, + test_user: User, + og_create_res_fact, + captured_query_events: list[dict], + ): + session = seeded_integration_session first = await resource_actions.create( session, test_user, og_create_res_fact(name="Merge A") ) second = await resource_actions.create( session, test_user, og_create_res_fact(name="Merge B") ) - # create (with source_data) fans out to the source helpers too. - assert { - "resources.create", - "sources.get_or_create", - "sources.attach", - } <= _query_names(captured_query_events) + await session.commit() captured_query_events.clear() await resource_actions.apply_resource_merge( session, test_user, [first.id, second.id] ) - assert { - "resources.merge.load", - "resources.merge.apply", - } <= _query_names(captured_query_events) + _assert_labels( + captured_query_events, {"resources.merge.load", "resources.merge.apply"} + ) @pytest.mark.anyio async def test_set_field_source_priority_labels( @@ -233,11 +267,18 @@ async def test_set_field_source_priority_labels( ) await fresh.commit() - assert { - "resources.set_field_source_priority.load", - "resources.set_field_source_priority.candidates", - "resources.set_field_source_priority.persist", - } <= _query_names(captured_query_events) + # The action re-prioritizes, then returns field_source_values(), which + # itself resolves the root and re-reads the candidates. + _assert_labels( + captured_query_events, + { + "resources.set_field_source_priority.load", + "resources.set_field_source_priority.candidates", + "resources.set_field_source_priority.persist", + "resources.resolve_root", + "resources.field_source_values", + }, + ) @pytest.mark.anyio async def test_source_action_labels( @@ -258,30 +299,40 @@ async def test_source_action_labels( created = await source_actions.create_source( session, test_user, source_maker(managed=False, source="gem") ) + _assert_labels(captured_query_events, {"sources.create"}) + + captured_query_events.clear() await source_actions.create_and_attach_sources( session, test_user, [source_maker(managed=False, source="rmi")], parent.id ) + _assert_labels(captured_query_events, {"sources.create_and_attach"}) + + captured_query_events.clear() await source_actions.get_or_create_sources( session, test_user, [source_maker(managed=False, source="wm")] ) + _assert_labels(captured_query_events, {"sources.get_or_create"}) + + captured_query_events.clear() await source_actions.attach_sources_to_resource( session, parent.id, [source_maker(managed=False, source="bc")], test_user ) + _assert_labels(captured_query_events, {"sources.attach"}) + + captured_query_events.clear() await source_actions.get_source(session, created.id) + _assert_labels(captured_query_events, {"sources.detail"}) + + captured_query_events.clear() await source_actions.get_sources(session, [created.id]) - await source_actions.query(session, OGFieldQueryParams()) + _assert_labels(captured_query_events, {"sources.get_by_ids"}) - assert { - "sources.create", - "sources.create_and_attach", - "sources.get_or_create", - "sources.attach", - "sources.detail", - "sources.get_by_ids", - "sources.count", - "sources.list_ids", - "sources.list_hydrate", - } <= _query_names(captured_query_events) + captured_query_events.clear() + await source_actions.query(session, OGFieldQueryParams()) + _assert_labels( + captured_query_events, + {"sources.count", "sources.list_ids", "sources.list_hydrate"}, + ) @pytest.mark.anyio async def test_merge_candidate_action_labels( @@ -308,15 +359,50 @@ async def test_merge_candidate_action_labels( resource_ids=[resources[0].id, resources[1].id] ), ) + _assert_labels( + captured_query_events, + { + "merge_candidates.create.load_resources", + "merge_candidates.create.check_existing", + "merge_candidates.create.persist", + }, + ) + + captured_query_events.clear() await mca.list_merge_candidates(session) + _assert_labels(captured_query_events, {"merge_candidates.list"}) + + captured_query_events.clear() await mca.get_merge_candidate(session, approved.id) + _assert_labels( + captured_query_events, + { + "merge_candidates.detail.load", + "merge_candidates.detail.coalesce", + "merge_candidates.detail.default_priority", + }, + ) + + captured_query_events.clear() await mca.approve_merge_candidate( session, test_user, approved.id, MergeCandidateReviewRequest(review_notes="ok"), ) + # approve delegates the actual merge to apply_resource_merge. + _assert_labels( + captured_query_events, + { + "merge_candidates.approve.load", + "merge_candidates.approve.load_resources", + "merge_candidates.approve.persist", + "resources.merge.load", + "resources.merge.apply", + }, + ) + captured_query_events.clear() denied = await mca.create_merge_candidate( session, test_user, @@ -330,21 +416,13 @@ async def test_merge_candidate_action_labels( denied.id, MergeCandidateReviewRequest(review_notes="no"), ) - - assert { - "merge_candidates.create.load_resources", - "merge_candidates.create.check_existing", - "merge_candidates.create.persist", - "merge_candidates.list", - "merge_candidates.detail.load", - "merge_candidates.detail.coalesce", - "merge_candidates.detail.default_priority", - "merge_candidates.approve.load", - "merge_candidates.approve.load_resources", - "merge_candidates.approve.persist", - "merge_candidates.deny.load", - "merge_candidates.deny.persist", - # approve delegates to apply_resource_merge - "resources.merge.load", - "resources.merge.apply", - } <= _query_names(captured_query_events) + _assert_labels( + captured_query_events, + { + "merge_candidates.create.load_resources", + "merge_candidates.create.check_existing", + "merge_candidates.create.persist", + "merge_candidates.deny.load", + "merge_candidates.deny.persist", + }, + )