From 7db8c92053552cf29fd3100d7a75c9c4a3eb1390 Mon Sep 17 00:00:00 2001 From: Eric Larson Date: Fri, 2 Oct 2026 15:59:55 -0500 Subject: [PATCH] Speed up dashboard repository counts with grouped queries --- .../app/services/ogm_harvest/repository.py | 98 ++++++++++--------- .../services/test_ogm_harvest_repository.py | 83 ++++++++++++++++ 2 files changed, 137 insertions(+), 44 deletions(-) diff --git a/backend/app/services/ogm_harvest/repository.py b/backend/app/services/ogm_harvest/repository.py index f82ece4..31ff877 100644 --- a/backend/app/services/ogm_harvest/repository.py +++ b/backend/app/services/ogm_harvest/repository.py @@ -3,7 +3,7 @@ from datetime import datetime from typing import Any, Dict, List, Optional -from sqlalchemy import String, cast, func, literal, select, update +from sqlalchemy import String, cast, func, literal, select, true, update from sqlalchemy.dialects.postgresql import insert as pg_insert from db.database import database @@ -40,12 +40,14 @@ async def list_public_repo_summaries(self) -> List[Dict[str, Any]]: .group_by(ogm_harvest_runs.c.ogm_repo_name) .subquery() ) - harvested_record_count = ( - select(func.count()) - .select_from(ogm_resource_state) - .where(ogm_resource_state.c.ogm_repo_name == ogm_repos.c.ogm_repo_name) + harvested_counts = ( + select( + ogm_resource_state.c.ogm_repo_name, + func.count().label("harvested_record_count"), + ) .where(ogm_resource_state.c.ogm_missing_since.is_(None)) - .scalar_subquery() + .group_by(ogm_resource_state.c.ogm_repo_name) + .subquery("harvested_counts") ) empty_string = cast(literal(""), String()) published_state = cast(literal("published"), String()) @@ -57,41 +59,31 @@ async def list_public_repo_summaries(self) -> List[Dict[str, Any]]: published_state, ) ) - published_available_record_count = ( - select(func.count()) - .select_from(resources) - .where(resources.c.b1g_adminTags_sm.is_not(None)) - .where( - resources.c.b1g_adminTags_sm.any( - func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name) - ) - ) - .where(effective_publication_state == published_state) - .scalar_subquery() + # Expand each resource's tags once, then calculate all counts together. + # DISTINCT preserves ANY's one-match-per-resource behavior for duplicate tags. + repo_tags = ( + select(func.unnest(resources.c.b1g_adminTags_sm).label("repo_tag")) + .distinct() + .correlate(resources) + .lateral("repo_tags") ) - suppressed_record_count = ( - select(func.count()) - .select_from(resources) - .where(resources.c.b1g_adminTags_sm.is_not(None)) - .where( - resources.c.b1g_adminTags_sm.any( - func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name) - ) - ) - .where(func.coalesce(resources.c.gbl_suppressed_b, False).is_(True)) - .scalar_subquery() - ) - unpublished_record_count = ( - select(func.count()) - .select_from(resources) - .where(resources.c.b1g_adminTags_sm.is_not(None)) - .where( - resources.c.b1g_adminTags_sm.any( - func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name) - ) + resource_counts = ( + select( + repo_tags.c.repo_tag, + func.count() + .filter(effective_publication_state == published_state) + .label("available_record_count"), + func.count() + .filter(resources.c.gbl_suppressed_b.is_(True)) + .label("suppressed_record_count"), + func.count() + .filter(effective_publication_state != published_state) + .label("unpublished_record_count"), ) - .where(effective_publication_state != published_state) - .scalar_subquery() + .select_from(resources.join(repo_tags, true())) + .where(func.starts_with(repo_tags.c.repo_tag, ogm_repo_tag_prefix)) + .group_by(repo_tags.c.repo_tag) + .subquery("resource_counts") ) q = ( @@ -109,17 +101,35 @@ async def list_public_repo_summaries(self) -> List[Dict[str, Any]]: ogm_harvest_runs.c.ogm_completed_at.label("last_run_completed_at"), ogm_harvest_runs.c.ogm_status.label("last_run_status"), ogm_harvest_runs.c.ogm_stats_json.label("last_run_stats_json"), - harvested_record_count.label("harvested_record_count"), - published_available_record_count.label("available_record_count"), - suppressed_record_count.label("suppressed_record_count"), - unpublished_record_count.label("unpublished_record_count"), + func.coalesce(harvested_counts.c.harvested_record_count, 0).label( + "harvested_record_count" + ), + func.coalesce(resource_counts.c.available_record_count, 0).label( + "available_record_count" + ), + func.coalesce(resource_counts.c.suppressed_record_count, 0).label( + "suppressed_record_count" + ), + func.coalesce(resource_counts.c.unpublished_record_count, 0).label( + "unpublished_record_count" + ), ) .select_from( ogm_repos.outerjoin( latest_run_ids, ogm_repos.c.ogm_repo_name == latest_run_ids.c.repo_name - ).outerjoin( + ) + .outerjoin( ogm_harvest_runs, ogm_harvest_runs.c.ogm_id == latest_run_ids.c.latest_ogm_id ) + .outerjoin( + harvested_counts, + harvested_counts.c.ogm_repo_name == ogm_repos.c.ogm_repo_name, + ) + .outerjoin( + resource_counts, + resource_counts.c.repo_tag + == func.concat(ogm_repo_tag_prefix, ogm_repos.c.ogm_repo_name), + ) ) .order_by(ogm_repos.c.ogm_repo_name) ) diff --git a/backend/tests/services/test_ogm_harvest_repository.py b/backend/tests/services/test_ogm_harvest_repository.py index 81e8c44..509966f 100644 --- a/backend/tests/services/test_ogm_harvest_repository.py +++ b/backend/tests/services/test_ogm_harvest_repository.py @@ -33,3 +33,86 @@ async def fake_fetch_all(query): assert "nullif(resources.b1g_publication_state_s, CAST(" in sql assert "nullif(resources.publication_state, CAST(" in sql assert ") = CAST(" in sql + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_public_repo_counts_preserve_tag_and_publication_semantics( + monkeypatch, db_connection +): + """Exercise real PostgreSQL array/count behavior using connection-local tables.""" + import os + + import asyncpg + from sqlalchemy.schema import CreateTable + + from db.models import ogm_harvest_runs, ogm_repos, ogm_resource_state + + url = os.getenv("OGM_QUERY_TEST_DATABASE_URL") + if not url and db_connection is not None: + url = str(db_connection.url).replace("postgresql+asyncpg://", "postgresql://") + if not url: + pytest.skip("Set OGM_QUERY_TEST_DATABASE_URL to run isolated PostgreSQL query tests") + connection = await asyncpg.connect(url) + try: + for table in (ogm_repos, ogm_harvest_runs, ogm_resource_state): + ddl = str(CreateTable(table).compile(dialect=database._backend._dialect)) + await connection.execute(ddl.replace("CREATE TABLE", "CREATE TEMP TABLE", 1)) + await connection.execute(""" + CREATE TEMP TABLE resources ( + id text PRIMARY KEY, "b1g_adminTags_sm" varchar[], + b1g_publication_state_s varchar, publication_state varchar, + gbl_suppressed_b boolean + ); + INSERT INTO ogm_repos (ogm_repo_name, ogm_enabled) VALUES + ('alpha', true), ('beta', false), ('empty', true); + INSERT INTO ogm_resource_state + (ogm_repo_name, ogm_resource_id, ogm_missing_since) VALUES + ('alpha', 'active', NULL), ('alpha', 'missing', now()), + ('beta', 'active', NULL); + INSERT INTO ogm_harvest_runs + (ogm_id, ogm_repo_name, ogm_trigger, ogm_status) VALUES + (1, 'alpha', 'manual', 'failed'), (2, 'alpha', 'manual', 'success'); + """) + await connection.executemany( + """INSERT INTO resources VALUES ($1, $2, $3, $4, $5)""", + [ + ("duplicate", ["ogm_repo:alpha", "ogm_repo:alpha"], None, None, None), + ("shared", ["ogm_repo:alpha", "ogm_repo:beta"], "PUBLISHED", "draft", True), + ("fallback", ["ogm_repo:alpha"], "", "Published", False), + ("unpublished", ["ogm_repo:alpha"], "Draft", "published", True), + ("legacy", ["ogm_repo:alpha"], None, "draft", None), + ("blank", ["ogm_repo:beta"], "", "", False), + ("null-tags", None, None, None, True), + ("empty-tags", [], None, None, True), + ("other-tags", [None, "ogmXrepo:alpha", "ogm_repo:unknown"], None, None, True), + ], + ) + + async def fetch_all(query): + compiled = query.compile( + dialect=database._backend._dialect, + compile_kwargs={"render_postcompile": True}, + ) + parameters = [value for _, value in sorted(compiled.params.items())] + return await connection.fetch(_compile_like_databases(query), *parameters) + + monkeypatch.setattr(ogm_repository.database, "fetch_all", fetch_all) + summaries = await OGMHarvestRepository().list_public_repo_summaries() + assert [row["ogm_repo_name"] for row in summaries] == ["alpha", "beta", "empty"] + counts = { + row["ogm_repo_name"]: ( + row["harvested_record_count"], + row["available_record_count"], + row["suppressed_record_count"], + row["unpublished_record_count"], + ) + for row in summaries + } + assert counts == {"alpha": (1, 3, 2, 2), "beta": (1, 2, 1, 0), "empty": (0, 0, 0, 0)} + assert summaries[0]["last_run_id"] == 2 + assert summaries[0]["last_crawl_status"] == "success" + assert summaries[1]["ogm_enabled"] is False + assert summaries[2]["last_run_id"] is None + finally: + await connection.close()