postgres-connect-crud

copy

import sys
import os
import subprocess
import time
import urllib.request
import json

# ─────────────────────────────────────────
# PRE_EXECUTION
# ─────────────────────────────────────────

# 1. Fetch registry entries via hard URLs
REGISTRY_URLS = [
    "https://checklist.day/api/registry/psycopg2-binary",
    "https://checklist.day/api/registry/sqlalchemy",
]

registry_data = {}
for url in REGISTRY_URLS:
    slug = url.split("/")[-1]
    for attempt in range(2):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "checklist-agent/1.0"})
            with urllib.request.urlopen(req, timeout=10) as resp:
                data = json.loads(resp.read())
                registry_data[slug] = data
                break
        except Exception as e:
            if attempt == 1:
                print(f"ABORT: registry unreachable for {slug} after 2 attempts — {e}")
                sys.exit(1)
            time.sleep(2)

# 2. Assert required fields
for slug, data in registry_data.items():
    for field in ["imports", "install", "warnings"]:
        if not data.get(field):
            print(f"ABORT: registry entry '{slug}' missing required field '{field}'")
            sys.exit(1)

# 3. Surface warnings
for slug, data in registry_data.items():
    warnings = data.get("warnings", [])
    if warnings:
        print(f"[{slug}] WARNINGS:")
        for w in warnings if isinstance(warnings, list) else [warnings]:
            print(f"  ⚠ {w}")

# ─────────────────────────────────────────
# EXECUTION
# ─────────────────────────────────────────

# 4. Auto-install deps
# FOOTGUN: install psycopg2-binary, NOT psycopg2
# psycopg2 requires libpq-dev + gcc build tools; psycopg2-binary is self-contained
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q",
                       "psycopg2-binary>=2.9.9", "sqlalchemy>=2.0.0"])

# 5. Correct imports per registry
# FOOTGUN: SQLAlchemy 2.x — use sqlalchemy.text() for raw SQL, not plain strings
# FOOTGUN: engine.execute() removed in SQLAlchemy 2.x — use engine.connect() + conn.execute()
# FOOTGUN: psycopg2 connect string uses host/dbname kwargs, not URL — use SQLAlchemy for URL parsing
from sqlalchemy import create_engine, text
from sqlalchemy.exc import OperationalError, ProgrammingError

# 6. Validate inputs
DATABASE_URL = os.environ.get("DATABASE_URL")
TABLE_NAME   = os.environ.get("TABLE_NAME", "checklist_test")

if not DATABASE_URL:
    print("ABORT: DATABASE_URL env var not set")
    print("  format: postgresql://user:pass@host:5432/dbname")
    sys.exit(1)

# FOOTGUN: SQLAlchemy 2.x requires postgresql:// not postgres://
# Heroku/Render/Supabase often give postgres:// URLs
if DATABASE_URL.startswith("postgres://"):
    DATABASE_URL = DATABASE_URL.replace("postgres://", "postgresql://", 1)
    print("  [fixed] postgres:// → postgresql:// (SQLAlchemy 2.x requirement)")

# 7. Connect
# FOOTGUN: use pool_pre_ping=True to detect stale connections
engine = create_engine(DATABASE_URL, pool_pre_ping=True)

conn = None
try:
    conn = engine.connect()

    # 8. Idempotent table create
    # FOOTGUN: always use IF NOT EXISTS — agents may retry
    conn.execute(text(f"""
        CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
            id SERIAL PRIMARY KEY,
            name TEXT NOT NULL,
            value INTEGER NOT NULL
        )
    """))
    conn.commit()

    # Clean slate for test — delete any existing test rows
    conn.execute(text(f"DELETE FROM {TABLE_NAME} WHERE name LIKE 'checklist_test_%'"))
    conn.commit()

    # 9. INSERT
    conn.execute(text(f"""
        INSERT INTO {TABLE_NAME} (name, value) VALUES
        ('checklist_test_alpha', 100),
        ('checklist_test_beta',  200)
    """))
    conn.commit()

    # 10. READ — verify insert
    # FOOTGUN: result.fetchall() must be called before conn.execute() again (cursor consumed)
    rows = conn.execute(text(f"SELECT name, value FROM {TABLE_NAME} WHERE name LIKE 'checklist_test_%' ORDER BY name")).fetchall()
    rows_inserted = len(rows)

    # 11. UPDATE
    conn.execute(text(f"UPDATE {TABLE_NAME} SET value = 999 WHERE name = 'checklist_test_alpha'"))
    conn.commit()

    updated = conn.execute(text(f"SELECT value FROM {TABLE_NAME} WHERE name = 'checklist_test_alpha'")).fetchone()
    update_ok = updated[0] == 999

    # 12. DELETE
    conn.execute(text(f"DELETE FROM {TABLE_NAME} WHERE name = 'checklist_test_beta'"))
    conn.commit()

    remaining = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE_NAME} WHERE name LIKE 'checklist_test_%'")).fetchone()
    delete_ok = remaining[0] == 1

    # Cleanup test rows
    conn.execute(text(f"DELETE FROM {TABLE_NAME} WHERE name LIKE 'checklist_test_%'"))
    conn.commit()

finally:
    if conn:
        conn.close()
    engine.dispose()

# ─────────────────────────────────────────
# POST_EXECUTION
# ─────────────────────────────────────────

# 13. Verify all operations
assert rows_inserted == 2, f"FAIL: expected 2 rows inserted, got {rows_inserted}"
assert rows[0][0] == "checklist_test_alpha" and rows[0][1] == 100, f"FAIL: read-back mismatch — {rows[0]}"
assert rows[1][0] == "checklist_test_beta"  and rows[1][1] == 200, f"FAIL: read-back mismatch — {rows[1]}"
assert update_ok, "FAIL: UPDATE did not persist"
assert delete_ok, "FAIL: DELETE row count wrong after delete"

# 14. Structured result
result = {
    "table_name":    TABLE_NAME,
    "rows_inserted": rows_inserted,
    "rows_read":     rows_inserted,
    "update_ok":     update_ok,
    "delete_ok":     delete_ok,
}
print(json.dumps(result, indent=2))
print("PASS")