Skip to content

Fix slow repository dashboard with grouped counts #16

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ewlarson merged 1 commit into from
Oct 2, 2026
Merged

Fix slow repository dashboard with grouped counts #16

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 54 additions & 44 deletions backend/app/services/ogm_harvest/repository.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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())
Expand All @@ -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 = (
Expand All @@ -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)
)
Expand Down
83 changes: 83 additions & 0 deletions backend/tests/services/test_ogm_harvest_repository.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Loading