fn_registry/docs/capabilities/postgres.md

Capability: postgres

CRUD de PostgreSQL desde el registry. Las funciones Python (psycopg2) reciben un dsn: str, son impuras y devuelven un dict {status:'ok'|'error', ...} sin lanzar (mismo estilo que el grupo duckdb); la función Go (postgres_open) abre un *sql.DB desde parámetros individuales.

Postgres es la capa que sirve datos a las herramientas de BI del stack (Excel → DuckDB → Postgres → visualización). Metabase, Grafana y Superset NO hablan DuckDB de forma nativa, pero todas hablan PostgreSQL: por eso el motor analítico de trabajo es DuckDB y, cuando un dashboard tiene que consumir esos datos, se sincronizan a Postgres con duckdb_to_postgres (grupo duckdb).

Funciones

ID	Firma	Que hace
postgres_open_go_infra	PostgresOpen(host, port, user, password, dbname, sslmode) (*sql.DB, error)	Conecta a PostgreSQL desde Go construyendo el DSN. sslmode por defecto disable.
pg_query_py_infra	pg_query(dsn, sql, params=None, max_rows=10000) -> dict	SELECT read-only (SET TRANSACTION READ ONLY) con RealDictCursor. Devuelve {status, columns, rows, row_count, truncated}. Normaliza tipos no JSON (date/datetime→ISO, Decimal→float, bytes→base64, UUID→str). Espejo de duckdb_query_readonly. Valores por %s.
pg_insert_rows_py_infra	pg_insert_rows(dsn, table, rows, add_snapshot_date=True) -> int	INSERT append-only en lote (execute_values). Deriva columnas de las claves. Opcional snapshot_date = date.today(). Retorna nº de filas.
pg_upsert_py_infra	pg_upsert(dsn, table, rows, key_cols, update_cols=None) -> dict	UPSERT idempotente INSERT ... ON CONFLICT (key_cols) DO UPDATE SET col=EXCLUDED.col. update_cols = ownership selectivo (las no listadas conservan su valor); [] = DO NOTHING. Devuelve {status, inserted, updated}. key_cols deben tener PK/UNIQUE. Espejo de duckdb_upsert.
pg_create_table_from_rows_py_infra	pg_create_table_from_rows(dsn, table, rows, primary_key=None) -> dict	CREATE TABLE IF NOT EXISTS infiriendo columnas y tipos desde los valores (bool→BOOLEAN, int→BIGINT, float→DOUBLE PRECISION, datetime→TIMESTAMP, date→DATE, resto→TEXT). Idempotente. Devuelve {status, created, table, columns}.
pg_list_tables_py_infra	pg_list_tables(dsn, schema='public') -> dict	Introspección read-only: tablas base con sus columnas vía information_schema. Devuelve {status, schema, tables:[{name, columns:[{name,type,nullable}]}]}.
pg_apply_sql_py_infra	pg_apply_sql(dsn, sql_path) -> int	Ejecuta un archivo .sql completo (multi-statement, una transacción). Para migraciones idempotentes (IF NOT EXISTS).

Relacionadas (otros grupos): duckdb_to_postgres_py_pipelines (sincroniza una tabla DuckDB a Postgres) e init_metabase_go_infra (despliega el stack Metabase + Postgres en Docker).

Ejemplo canónico

Crear una tabla inferida, hacer upsert idempotente y consultar (DSN desde pass):

cd /home/enmanuel/fn_registry
DSN="postgresql://captacion:$(pass captacion/postgres | head -1)@localhost:5433/trends"
python/.venv/bin/python3 - "$DSN" <<'PYEOF'
import sys
sys.path.insert(0, "python/functions")
from infra import pg_create_table_from_rows, pg_upsert, pg_query

dsn = sys.argv[1]
rows = [{"mes": "2026-01", "total": 12500.5}, {"mes": "2026-02", "total": 15800.75}]

pg_create_table_from_rows(dsn, "demo_kpi", rows, primary_key=["mes"])
print(pg_upsert(dsn, "demo_kpi", rows, key_cols=["mes"]))          # inserted/updated
print(pg_upsert(dsn, "demo_kpi", rows, key_cols=["mes"]))          # idempotente: 0 inserts
print(pg_query(dsn, "SELECT * FROM demo_kpi ORDER BY mes")["rows"])
PYEOF

Gotchas del grupo

• El DSN lleva credenciales — nunca hardcodear. Resuélvelo desde pass (ej. pass captacion/postgres: L1 = password, resto user/host/port/datadb). No imprimas el DSN en logs.
• pg_query/pg_list_tables son read-only por convención (SET TRANSACTION READ ONLY + rollback), protegen la base pero NO son sandbox; los identificadores (tabla/schema) NO se parametrizan — los valores sí (%s). Las funciones validan identificadores con ^[A-Za-z_][A-Za-z0-9_]*$.
• pg_upsert cuenta insert vs update con el pseudo-columna xmax (RETURNING (xmax = 0)). Fiable en el caso normal (single-writer, sin triggers raros). Con update_cols=[] (DO NOTHING) las filas en conflicto no se devuelven, así que solo se cuentan las nuevas. BEFORE-triggers / REPLICA IDENTITY pueden desviar el conteo.
• pg_create_table_from_rows no reconcilia schema: si la tabla ya existe, columns reporta los tipos inferidos de las filas, no los reales. Inferencia best-effort sin NUMERIC/escala — para dinero define el schema a mano con pg_apply_sql.
• pg_insert_rows y pg_apply_sql lanzan en error (no devuelven dict); envuélvelas si compones.

Fronteras

• NO es el motor analítico del stack — ese es DuckDB (columnar, lee CSV/Parquet/Excel nativo). Postgres es el destino para BI.
• NO dibuja dashboards: eso es Metabase / Grafana / Evidence leyendo de Postgres.
• NO cubre PostGIS más allá de osm2pgsql_ingest_py_infra (geo, aparte).

Relación con otros grupos

• duckdb — duckdb_to_postgres es el puente de entrada de datos a esta capa.
• metabase — registra la base con metabase_add_database(engine='postgres', ...) y consume las tablas.
• excel — el origen de los datos suele ser un .xlsx ingerido por excel_to_duckdb.