55 lines
2.3 KiB
Python
55 lines
2.3 KiB
Python
"""Pipeline one-shot: ejecuta un SELECT contra el Postgres de un proyecto conocido.
|
|
|
|
Compone dos funciones del registry sin reescribir su lógica:
|
|
1. resolve_pg_dsn(project) -> resuelve el DSN del proyecto (env / .env / pass).
|
|
2. pg_query(dsn, sql, max_rows=...) -> ejecuta el SELECT read-only y devuelve
|
|
las filas como list[dict].
|
|
|
|
Elimina el patrón inline que el agente repetía: resolver el DSN a mano y luego
|
|
lanzar psql/psycopg2 con él. El caller solo necesita el nombre del proyecto y
|
|
el SQL; el password sale de pass en runtime, nunca está hardcodeado.
|
|
|
|
Es un pipeline (kind: pipeline -> siempre impuro). Devuelve un dict sin lanzar:
|
|
lo que devuelve pg_query en éxito, o el error de resolución del DSN si falla
|
|
el primer paso.
|
|
"""
|
|
|
|
from infra.resolve_pg_dsn import resolve_pg_dsn
|
|
from infra.pg_query import pg_query
|
|
|
|
|
|
def query_project_pg(project: str, sql: str, max_rows: int = 10000) -> dict:
|
|
"""Resuelve el DSN de un proyecto y ejecuta un SELECT contra su Postgres.
|
|
|
|
Args:
|
|
project: nombre del proyecto conocido ('captacion' / 'captacion_clientes',
|
|
'seo' / 'seo_analytics'). Se pasa tal cual a resolve_pg_dsn.
|
|
sql: sentencia SQL a ejecutar (pensada para SELECT). Para parámetros, usa
|
|
el marcador %s; este pipeline no expone params posicionales, así que
|
|
interpola valores constantes y de confianza solo (para entradas no
|
|
confiables usa pg_query directamente con params).
|
|
max_rows: número máximo de filas a materializar (default 10000). Se pasa
|
|
tal cual a pg_query; si la query produce más, el resultado se trunca.
|
|
|
|
Returns:
|
|
dict. En éxito propaga el resultado de pg_query:
|
|
{status:'ok', columns, rows, row_count, truncated}. Si la resolución del
|
|
DSN falla, propaga {status:'error', error} de resolve_pg_dsn sin tocar
|
|
Postgres.
|
|
"""
|
|
resolved = resolve_pg_dsn(project)
|
|
if resolved.get("status") != "ok":
|
|
return resolved
|
|
return pg_query(resolved["dsn"], sql, max_rows=max_rows)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Demo lanzable: cuenta de oportunidades de producto en captacion_clientes.
|
|
import json
|
|
|
|
out = query_project_pg(
|
|
"captacion",
|
|
"SELECT COUNT(*) AS n FROM product_opportunities",
|
|
)
|
|
print(json.dumps(out, ensure_ascii=False))
|