Add psql \copy Postgres push flow, CLI support, and a manual-trigger GH Action

Replaces the dataflows-based dump_to_sql call with a simpler
psql \copy of the already-built CSV file - matches how a human would
push this by hand, and decouples the Postgres push from re-serializing
the whole dataset a second way. Truncates the target table first by
default (matching the old dump_to_sql(mode='rewrite') semantics), or
--no-truncate to append.

New CLI: `lamas push-postgres [--csv PATH] [--table T] [--truncate/
--no-truncate]`, usable standalone or via `lamas build --output
local,postgres`. Verified end-to-end against a disposable smoke-test
table on the real database (created, copied 5 rows in, verified,
dropped) before wiring anything to the production table.

Added .github/workflows/push-postgres.yml - workflow_dispatch only
(never runs on push/PR), requires typing "PUSH" to confirm, runs the
full QA suite before pushing. Uses the DATAFLOWS_DB_ENGINE repo secret
(set via `gh secret set`, never written to any file in this repo).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: akariv

.github/workflows/push-postgres.yml

@@ -0,0 +1,76 @@
+name: Push to Postgres
+
+# Manual trigger ONLY - this truncates and overwrites the production lamas_muni table. It must
+# never run automatically on push/PR; every other workflow in this repo pushes nothing.
+on:
+  workflow_dispatch:
+    inputs:
+      confirm:
+        description: 'Type PUSH to confirm this will TRUNCATE and overwrite the production lamas_muni table'
+        required: true
+
+jobs:
+  push:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Confirm intent
+        run: |
+          if [ "${{ github.event.inputs.confirm }}" != "PUSH" ]; then
+            echo "::error::Confirmation input must be exactly 'PUSH' - aborting."
+            exit 1
+          fi
+
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+
+      - name: Ensure psql is available
+        run: command -v psql || (sudo apt-get update && sudo apt-get install -y postgresql-client)
+
+      - name: Install lamas package
+        run: pip install -e "Lamas/[dev]"
+
+      # Same cache keys as lamas-tests.yml (repo-scoped, so this reuses whatever it already
+      # populated) - see that workflow for the rationale.
+      - name: Cache downloaded workbooks
+        uses: actions/cache@v4
+        with:
+          path: Lamas/downloads
+          key: lamas-downloads-v1
+
+      - name: Download all configured years
+        run: lamas download
+        continue-on-error: true
+
+      - name: Cache preprocessed checkpoint
+        uses: actions/cache@v4
+        with:
+          path: Lamas/.cache
+          key: lamas-checkpoint-${{ hashFiles('Lamas/downloads/**', 'Lamas/lamas/preprocess.py', 'Lamas/lamas/sheet_config.py', 'Lamas/data/sheet_config.yaml') }}
+
+      - name: Ensure data is available (checkpoint or fresh download)
+        run: |
+          if [ -f Lamas/.cache/preprocessed.parquet ]; then
+            echo "Using cached checkpoint"
+          elif [ -n "$(ls -A Lamas/downloads 2>/dev/null)" ]; then
+            lamas preprocess
+          else
+            echo "::error::No downloaded workbooks and no cached checkpoint - cannot build"
+            exit 1
+          fi
+
+      - name: Regenerate pending-headers report
+        run: lamas map-headers
+
+      - name: Run the full QA suite before pushing
+        run: pytest Lamas/tests/ -v
+
+      - name: Build (local output, strict - refuses if any header is unresolved)
+        run: lamas build --strict --output local
+
+      - name: Push to Postgres
+        env:
+          DATAFLOWS_DB_ENGINE: ${{ secrets.DATAFLOWS_DB_ENGINE }}
+        run: lamas push-postgres

Lamas/README.md

@@ -43,12 +43,28 @@ drop into just the step you need while iterating on a new year.
 | `lamas mapping confirm-fuzzy --orig "..."` | Accept the fuzzy-match suggestion already recorded for a pending row in `pending_headers.csv`. |
 | `lamas mapping reject-fuzzy --orig "..." [--canonical "..."]` | Override a fuzzy suggestion: map to a different canonical, or omit `--canonical` to make it a brand-new one. |
 | `lamas stats [--year Y]` | Regenerate `reports/header_stats.{csv,md}` - per-header row counts and year coverage, the local replacement for eyeballing the old Airtable "Stats" table. |
-| `lamas build [--strict] [--output local,postgres] [--output-dir PATH]` | Apply `specific_fixes`/`value_fixes` and write the final output. `--strict` refuses to build while any header is unresolved. Defaults to local-only output. |
+| `lamas build [--strict] [--output local,postgres] [--output-dir PATH]` | Apply `specific_fixes`/`value_fixes` and write the final output. `--strict` refuses to build while any header is unresolved. Defaults to local-only output; `postgres` pushes the resulting CSV via `push-postgres` below. |
+| `lamas push-postgres [--csv PATH] [--table lamas_muni] [--truncate/--no-truncate]` | Push an already-built CSV (default `db_bkp/res_1.csv`) into Postgres via `psql \copy`. Reads the connection string from the `DATAFLOWS_DB_ENGINE` env var. Truncates the table first by default (matching a full-replace, not an incremental append) - a destructive, shared-system write, never run automatically. |
 | `lamas full-run [--year Y] [--strict] [--output ...]` | `download` -> `preprocess` -> `map-headers` -> `stats` -> `build`, one shot. |
 | `lamas qa` | Runs the full pytest suite under `tests/` - the automated quality gate (see below). |
 
 Run `lamas <command> --help` for the full flag list on anything above.
 
+### Pushing to Postgres
+
+`DATAFLOWS_DB_ENGINE` is not auto-loaded from `.env` (nothing in the package calls
+`load_dotenv()`) - export it into your shell first:
+```bash
+set -a; source Lamas/.env; set +a
+```
+Then either let `build` push it as part of the pipeline (`lamas build --strict --output local,postgres`),
+or push a CSV you already have on hand directly:
+```bash
+lamas push-postgres --csv db_bkp/res_1.csv
+```
+A manually-triggered GitHub Action (`.github/workflows/push-postgres.yml`) runs this same flow
+using a repo secret - see that workflow file for details. It is never triggered automatically.
+
 ### Quality assurance
 
 `lamas qa` (or plain `pytest tests/`) runs the same checks a human used to do by eye in Airtable's

Lamas/lamas/cli.py

@@ -15,7 +15,7 @@
 from .headers.specific_fixes import specific_fixes
 from .headers.value_fixes import value_fixes
 from .io.local import write_local
-from .io.postgres import PostgresNotConfiguredError, write_postgres
+from .io.postgres import PostgresNotConfiguredError, push_csv_to_postgres
 from .logging_config import setup_logging
 from .sheet_config import SheetConfig
 from .stats import compute_header_stats, write_stats_report
@@ -299,15 +299,36 @@ def build(checkpoint, mapping_path, strict, output, output_dir):
             f'{len(unresolved)} unresolved headers remain - refusing to build in --strict mode'
         )
     targets = [t.strip() for t in output.split(',') if t.strip()]
-    if 'local' in targets:
+    # Postgres is pushed from the local CSV (via psql \copy - see push_csv_to_postgres), so the
+    # local write always happens first when postgres is requested, even if 'local' wasn't asked
+    # for explicitly.
+    if 'local' in targets or 'postgres' in targets:
         write_local(resolved_rows, output_dir)
         click.echo(f'Wrote local output to {output_dir}')
     if 'postgres' in targets:
+        csv_path = Path(output_dir) / 'res_1.csv'
         try:
-            write_postgres(resolved_rows)
+            push_csv_to_postgres(csv_path)
         except PostgresNotConfiguredError as e:
             raise click.ClickException(str(e))
-        click.echo('Wrote to Postgres table lamas_muni')
+        click.echo('Pushed to Postgres table lamas_muni')
+
+
+@main.command('push-postgres')
+@click.option('--csv', 'csv_path', type=click.Path(exists=True), default=str(DEFAULT_OUTPUT_DIR / 'res_1.csv'))
+@click.option('--table', default='lamas_muni')
+@click.option('--truncate/--no-truncate', default=True, help='Truncate the table before copying in (default: on).')
+def push_postgres(csv_path, table, truncate):
+    """Push an already-built CSV file (see `lamas build --output local`) into Postgres via `psql \\copy`.
+
+    Reads the connection string from the DATAFLOWS_DB_ENGINE environment variable. This is a
+    shared-system, destructive write (truncates the table by default) - never run automatically.
+    """
+    try:
+        push_csv_to_postgres(csv_path, table=table, truncate=truncate)
+    except PostgresNotConfiguredError as e:
+        raise click.ClickException(str(e))
+    click.echo(f'Pushed {csv_path} -> Postgres table {table!r}')
 
 
 @main.command('full-run')

Lamas/lamas/io/postgres.py

@@ -1,26 +1,30 @@
 import os
+import re
+import subprocess
+from pathlib import Path
 
-import dataflows as DF
+TABLE_IDENTIFIER = re.compile(r'^[A-Za-z_][A-Za-z0-9_]*$')
 
 
 class PostgresNotConfiguredError(Exception):
     pass
 
 
-def write_postgres(rows, table='lamas_muni', mode='rewrite'):
-    """Write the final dataset to Postgres. This formalizes the DF.dump_to_sql(...) call that was
-    commented out in the original notebook - it is only ever invoked when the user explicitly
-    asks for `--output postgres`, never by default."""
-    engine = os.environ.get('DATAFLOWS_DB_ENGINE')
+def push_csv_to_postgres(csv_path, table='lamas_muni', truncate=True, engine=None):
+    """Push an already-built CSV file (see `write_local` / `lamas build --output local`) into
+    Postgres via `psql \\copy` - simpler than staging the whole dataset through dataflows'
+    dump_to_sql, and matches how a human would do this by hand. `truncate=True` (the default)
+    replaces the table's contents, matching the previous dump_to_sql(mode='rewrite') behavior -
+    otherwise re-running this would just keep appending duplicate rows."""
+    if not TABLE_IDENTIFIER.match(table):
+        raise ValueError(f'Invalid table name: {table!r}')
+    engine = engine or os.environ.get('DATAFLOWS_DB_ENGINE')
     if not engine:
         raise PostgresNotConfiguredError('DATAFLOWS_DB_ENGINE is not set - cannot write to Postgres')
-    DF.Flow(
-        rows,
-        DF.update_resource(-1, name='lamas'),
-        DF.dump_to_sql({
-            table: {
-                'resource-name': 'lamas',
-                'mode': mode,
-            }
-        }, engine=engine, batch_size=1000),
-    ).process()
+    csv_path = str(Path(csv_path).resolve())
+    escaped_path = csv_path.replace("'", "''")
+    commands = []
+    if truncate:
+        commands += ['-c', f'TRUNCATE TABLE {table};']
+    commands += ['-c', f"\\copy {table} FROM '{escaped_path}' WITH (FORMAT csv, HEADER true)"]
+    subprocess.run(['psql', engine, '-v', 'ON_ERROR_STOP=1', *commands], check=True)

Lamas/tests/test_postgres.py

@@ -0,0 +1,48 @@
+import pytest
+
+from lamas.io.postgres import PostgresNotConfiguredError, push_csv_to_postgres
+
+
+def test_raises_when_engine_not_configured(monkeypatch, tmp_path):
+    monkeypatch.delenv('DATAFLOWS_DB_ENGINE', raising=False)
+    csv_path = tmp_path / 'res_1.csv'
+    csv_path.write_text('year,name\n2024,foo\n')
+    with pytest.raises(PostgresNotConfiguredError):
+        push_csv_to_postgres(str(csv_path))
+
+
+def test_rejects_unsafe_table_name(tmp_path):
+    csv_path = tmp_path / 'res_1.csv'
+    csv_path.write_text('year,name\n2024,foo\n')
+    with pytest.raises(ValueError):
+        push_csv_to_postgres(str(csv_path), table='lamas_muni; DROP TABLE foo;', engine='postgresql://x')
+
+
+def test_invokes_psql_with_truncate_and_copy(monkeypatch, tmp_path):
+    csv_path = tmp_path / 'res_1.csv'
+    csv_path.write_text('year,name\n2024,foo\n')
+    calls = []
+    monkeypatch.setattr('lamas.io.postgres.subprocess.run', lambda args, **kw: calls.append(args))
+
+    push_csv_to_postgres(str(csv_path), table='lamas_muni', engine='postgresql://x')
+
+    assert calls
+    args = calls[0]
+    assert args[0] == 'psql'
+    assert args[1] == 'postgresql://x'
+    joined = ' '.join(args)
+    assert 'TRUNCATE TABLE lamas_muni;' in joined
+    assert f"\\copy lamas_muni FROM '{csv_path.resolve()}' WITH (FORMAT csv, HEADER true)" in joined
+
+
+def test_no_truncate_when_disabled(monkeypatch, tmp_path):
+    csv_path = tmp_path / 'res_1.csv'
+    csv_path.write_text('year,name\n2024,foo\n')
+    calls = []
+    monkeypatch.setattr('lamas.io.postgres.subprocess.run', lambda args, **kw: calls.append(args))
+
+    push_csv_to_postgres(str(csv_path), table='lamas_muni', truncate=False, engine='postgresql://x')
+
+    joined = ' '.join(calls[0])
+    assert 'TRUNCATE' not in joined
+    assert '\\copy lamas_muni FROM' in joined