feat(db): persist each import's row counts (#89)

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}")

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))

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()
             ]

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:

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.

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

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))

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) '