postgres-json-column-query

copy

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

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

for attempt in range(2):
    try:
        req = urllib.request.Request(
            "https://checklist.day/api/registry/psycopg2-binary",
            headers={"User-Agent": "checklist-agent/1.0"}
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            registry = pyjson.loads(resp.read())
            warnings = registry.get("warnings", [])
            if warnings:
                print("[psycopg2-binary] WARNINGS:")
                for w in warnings if isinstance(warnings, list) else [warnings]:
                    print(f"  ⚠ {w}")
        break
    except Exception as e:
        if attempt == 1:
            print(f"ABORT: registry unreachable — {e}")
            sys.exit(1)
        time.sleep(2)

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

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "psycopg2-binary>=2.9.9"])

import psycopg2
import psycopg2.extras

DATABASE_URL = os.environ.get("DATABASE_URL")
if not DATABASE_URL:
    print("ABORT: DATABASE_URL env var not set")
    sys.exit(1)

if DATABASE_URL.startswith("postgres://"):
    DATABASE_URL = DATABASE_URL.replace("postgres://", "postgresql://", 1)

TABLE = "checklist_jsonb_test"
conn = psycopg2.connect(DATABASE_URL, connect_timeout=10)

try:
    with conn.cursor() as cur:
        # Setup
        cur.execute(f"""
            CREATE TABLE IF NOT EXISTS {TABLE} (
                id   SERIAL PRIMARY KEY,
                data JSONB NOT NULL
            )
        """)
        # GIN index for fast JSONB queries
        cur.execute(f"CREATE INDEX IF NOT EXISTS idx_{TABLE}_data ON {TABLE} USING GIN (data)")
        cur.execute(f"DELETE FROM {TABLE}")

        # FOOTGUN: use psycopg2.extras.Json to insert JSONB, not plain string
        records = [
            {"name": "alice", "score": 90, "tags": ["admin", "user"]},
            {"name": "bob",   "score": 75, "tags": ["user"]},
        ]
        for rec in records:
            cur.execute(
                f"INSERT INTO {TABLE} (data) VALUES (%s)",
                (psycopg2.extras.Json(rec),)
            )
        conn.commit()

        # 1. -> returns JSONB (object/array), ->> returns TEXT
        # FOOTGUN: comparing -> result to a string always fails — must use ->>
        cur.execute(f"SELECT pg_typeof(data -> 'name'), pg_typeof(data ->> 'name') FROM {TABLE} LIMIT 1")
        arrow_type, double_arrow_type = cur.fetchone()
        arrow_returns_jsonb = arrow_type == "jsonb"
        double_arrow_returns_text = double_arrow_type == "text"
        print(f"  -> returns: {arrow_type} (expected jsonb)")
        print(f"  ->> returns: {double_arrow_type} (expected text)")

        # 2. Correct way to filter by JSONB text value — use ->>
        cur.execute(f"SELECT data->>'name' FROM {TABLE} WHERE data->>'name' = %s", ("alice",))
        found = cur.fetchone()
        print(f"  ->> filter found: {found[0] if found else None}")

        # 3. @> containment operator
        cur.execute(
            f"SELECT COUNT(*) FROM {TABLE} WHERE data @> %s",
            (psycopg2.extras.Json({"tags": ["admin"]}),)
        )
        contains_count = cur.fetchone()[0]
        contains_operator_ok = contains_count == 1
        print(f"  @> containment: {contains_count} match (expected 1)")

        # 4. Verify GIN index exists
        cur.execute(f"""
            SELECT COUNT(*) FROM pg_indexes
            WHERE tablename = '{TABLE}' AND indexname = 'idx_{TABLE}_data'
        """)
        index_used = cur.fetchone()[0] == 1
        print(f"  GIN index exists: {index_used}")

        # Cleanup
        cur.execute(f"DROP TABLE IF EXISTS {TABLE}")
        conn.commit()

finally:
    conn.close()

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

assert arrow_returns_jsonb, "FAIL: -> should return jsonb"
assert double_arrow_returns_text, "FAIL: ->> should return text"
assert contains_operator_ok, f"FAIL: @> containment returned {contains_count}, expected 1"
assert index_used, "FAIL: GIN index was not created"

result = {
    "arrow_returns_jsonb":       arrow_returns_jsonb,
    "double_arrow_returns_text": double_arrow_returns_text,
    "contains_operator_ok":      contains_operator_ok,
    "index_used":                index_used,
}
print(pyjson.dumps(result, indent=2))
print("PASS")