From 37d32444221eeac2724954f465c661b8d61a9fdb Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 16:57:28 +1100 Subject: [PATCH] feat(db): persist each import's row counts on its registry row (BR-34 part) Migration 0003 adds total_rows, duplicate_rows_skipped and failed_rows to csv_files (row_count already holds the inserted count). Both loaders write them at registration and GET /api/csv/files returns them. Rows registered before 0003 keep NULL: their counts were never recorded. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019YPoXzBSj89ZHgw7nfTmbB --- alembic/versions/0003_import_counts.py | 38 ++++++++++++++ api/db.py | 5 +- api/routers/csv_routes.py | 6 ++- api/services/dynamic_loader.py | 8 +-- api/services/te_loader.py | 8 +-- scripts/lint.sh | 2 + tests/test_import_counts.py | 70 ++++++++++++++++++++++++++ tests/test_migrations.py | 2 +- 8 files changed, 130 insertions(+), 9 deletions(-) create mode 100644 alembic/versions/0003_import_counts.py create mode 100644 tests/test_import_counts.py diff --git a/alembic/versions/0003_import_counts.py b/alembic/versions/0003_import_counts.py new file mode 100644 index 0000000..feda44e --- /dev/null +++ b/alembic/versions/0003_import_counts.py @@ -0,0 +1,38 @@ +"""Persist each import's row counts on its registry row (BR-34, part). + +Until now only the inserted count (row_count) was stored; total, duplicate +and failed counts were returned to the caller and then lost. Rows registered +before this revision keep NULL in the new columns: their counts were never +recorded, and NULL says so rather than inventing a zero. + +Revision ID: 0003 +Revises: 0002 +Create Date: 2026-10-26 +""" + +from alembic import op + +from api.config import settings + +revision = "0003" +down_revision = "0002" +branch_labels = None +depends_on = None + +_COLUMNS = ("total_rows", "duplicate_rows_skipped", "failed_rows") + + +def _schema() -> str: + return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA) + + +def upgrade() -> None: + s = _schema() + for c in _COLUMNS: + op.execute(f"ALTER TABLE {s}.csv_files ADD COLUMN IF NOT EXISTS {c} INTEGER") + + +def downgrade() -> None: + s = _schema() + for c in _COLUMNS: + op.execute(f"ALTER TABLE {s}.csv_files DROP COLUMN IF EXISTS {c}") diff --git a/api/db.py b/api/db.py index 6dcdbed..f681d2d 100644 --- a/api/db.py +++ b/api/db.py @@ -204,7 +204,10 @@ def _bootstrap_fallback_schema() -> None: mode TEXT NOT NULL, row_count INTEGER NOT NULL DEFAULT 0, column_names TEXT[] NOT NULL DEFAULT '{{}}', - created_at TIMESTAMPTZ NOT NULL DEFAULT now() + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + total_rows INTEGER, + duplicate_rows_skipped INTEGER, + failed_rows INTEGER ) """ ).format(sql.Identifier(schema)) diff --git a/api/routers/csv_routes.py b/api/routers/csv_routes.py index baa4d86..af35666 100644 --- a/api/routers/csv_routes.py +++ b/api/routers/csv_routes.py @@ -100,7 +100,8 @@ def list_files() -> list[dict]: with conn.cursor() as cur: cur.execute( sql.SQL( - "SELECT id, file_name, table_name, mode, row_count, column_names, created_at " + "SELECT id, file_name, table_name, mode, row_count, column_names, created_at, " + "total_rows, duplicate_rows_skipped, failed_rows " "FROM {}.csv_files ORDER BY created_at DESC, id DESC" ).format(sql.Identifier(settings.UPLOADS_SCHEMA)) ) @@ -113,6 +114,9 @@ def list_files() -> list[dict]: "row_count": r[4], "column_names": r[5], "created_at": r[6].isoformat(), + "total_rows": r[7], + "duplicate_rows_skipped": r[8], + "failed_rows": r[9], } for r in cur.fetchall() ] diff --git a/api/services/dynamic_loader.py b/api/services/dynamic_loader.py index 7b1d457..2949f68 100644 --- a/api/services/dynamic_loader.py +++ b/api/services/dynamic_loader.py @@ -340,10 +340,12 @@ def _do_upload( try: cur.execute( sql.SQL( - "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names) " - "VALUES (%s, %s, %s, 'dynamic', %s, %s) RETURNING id" + "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names, " + "total_rows, duplicate_rows_skipped, failed_rows) " + "VALUES (%s, %s, %s, 'dynamic', %s, %s, %s, %s, %s) RETURNING id" ).format(sql.Identifier(schema)), - (file_name, file_hash, table_name, inserted, columns), + (file_name, file_hash, table_name, inserted, columns, + len(data_rows), duplicates, failed_row_count), ) file_id = cur.fetchone()[0] except psycopg2.errors.UniqueViolation: diff --git a/api/services/te_loader.py b/api/services/te_loader.py index d9386f2..8c500dc 100644 --- a/api/services/te_loader.py +++ b/api/services/te_loader.py @@ -294,10 +294,12 @@ def _err(row_number: int, column: str, value: str, reason: str) -> None: try: cur.execute( sql.SQL( - "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names) " - "VALUES (%s, %s, %s, 'te', %s, %s) RETURNING id" + "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names, " + "total_rows, duplicate_rows_skipped, failed_rows) " + "VALUES (%s, %s, %s, 'te', %s, %s, %s, 0, %s) RETURNING id" ).format(sql.Identifier(settings.UPLOADS_SCHEMA)), - (file_name, file_hash, f"{settings.TE_SCHEMA}.{target_table}", inserted, columns), + (file_name, file_hash, f"{settings.TE_SCHEMA}.{target_table}", inserted, columns, + len(data_rows), failed_row_count), ) except psycopg2.errors.UniqueViolation: # #81: a concurrent load of the same file registered first. diff --git a/scripts/lint.sh b/scripts/lint.sh index a11bf4e..a83313c 100644 --- a/scripts/lint.sh +++ b/scripts/lint.sh @@ -16,6 +16,7 @@ PY_SRC=( alembic/env.py alembic/versions/0001_baseline.py alembic/versions/0002_registry_unique.py + alembic/versions/0003_import_counts.py ) PY_TESTS=( @@ -28,6 +29,7 @@ PY_TESTS=( tests/test_api_hardening.py tests/test_evals_tier_e_x.py tests/test_migrations.py + tests/test_import_counts.py ) # Collect scripts/*.py dynamically diff --git a/tests/test_import_counts.py b/tests/test_import_counts.py new file mode 100644 index 0000000..ed12fd1 --- /dev/null +++ b/tests/test_import_counts.py @@ -0,0 +1,70 @@ +"""Each import's row counts are persisted on its registry row (BR-34, part). + +Integration: needs a live PostgreSQL. Fails (never skips) without one. +""" + +from __future__ import annotations + +import os +import unittest +import uuid + +import pytest +from fastapi.testclient import TestClient + +from api.main import app + +_HELP = ("Start PostgreSQL and ensure api/ can reach it (PGHOST/PGPORT/PGUSER/" + "PGPASSWORD/PGDATABASE). See scripts/start-api.ps1.") + + +@pytest.mark.integration +class ImportCountsArePersisted(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.ctx = TestClient(app) + try: + cls.client = cls.ctx.__enter__() + except Exception as exc: # noqa: BLE001 + raise AssertionError(f"Could not start the API: {type(exc).__name__}: {exc}\n{_HELP}") from None + + @classmethod + def tearDownClass(cls): + cls.ctx.__exit__(None, None, None) + + def setUp(self): + self.tag = f"{os.getpid()}_{uuid.uuid4().hex[:8]}" + + def _registry_entry(self, file_id: str) -> dict: + files = self.client.get("/api/csv/files").json() + return next(f for f in files if f["id"] == file_id) + + def _assert_persisted_matches_response(self, res: dict) -> dict: + entry = self._registry_entry(res["fileId"]) + self.assertEqual( + (entry["total_rows"], entry["row_count"], entry["duplicate_rows_skipped"], entry["failed_rows"]), + (res["totalRows"], res["insertedRows"], res["duplicateRowsSkipped"], res["failedRows"]), + ) + return entry + + def test_dynamic_import_persists_all_four_counts(self): + # 5 data rows: 2 valid distinct, 1 in-file duplicate, 2 that fail the int8 cast. + content = f"n,t\n1,{self.tag}\n2,{self.tag}\n1,{self.tag}\nx,{self.tag}\ny,{self.tag}\n" + res = self.client.post("/api/csv/upload", json={ + "fileName": f"counts_{self.tag}.csv", "content": content, "types": ["int8", "text"], + }).json() + self.assertEqual(res["status"], "ok", res) + entry = self._assert_persisted_matches_response(res) + self.assertEqual((entry["total_rows"], entry["row_count"], entry["duplicate_rows_skipped"], + entry["failed_rows"]), (5, 2, 1, 2)) + + def test_te_import_persists_counts(self): + content = (f"name,org_type,country\nCNT_{self.tag}_1,government,AU\n" + f"CNT_{self.tag}_2,government,AU\n") + res = self.client.post("/api/csv/upload", json={ + "fileName": f"counts_te_{self.tag}.csv", "content": content, + "mode": "te", "targetTable": "organisations", + }).json() + self.assertEqual(res["status"], "ok", res) + entry = self._assert_persisted_matches_response(res) + self.assertEqual((entry["total_rows"], entry["row_count"], entry["failed_rows"]), (2, 2, 0)) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 61648c3..87bcb6c 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -134,7 +134,7 @@ def setUp(self): def test_bootstrap_brings_the_schema_to_head(self): with Conn() as conn, conn.cursor() as cur: cur.execute(f'SELECT version_num FROM "{settings.UPLOADS_SCHEMA}".alembic_version') - self.assertEqual(cur.fetchone()[0], "0002") + self.assertEqual(cur.fetchone()[0], "0003") def test_duplicate_hash_is_rejected_by_the_database(self): insert = (f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files (file_name, file_hash, table_name, mode) '