Create and refactor Python SQL data access using psycopg and psycopg_pool only. Use for connection-info builders, schema-safe SQL setup, simple async helper functions, JSONB updates, and no-ORM patterns in gatekeeper services.
Produce small, explicit, async database functions that:
psycopg + psycopg_pool directly (no ORM).Use this skill when implementing or modifying DB access in this repository for:
Do not use this skill when:
Confirm required configuration inputs. Check all required env vars for DB auth and target DB. Fail fast with clear if any required var is missing.
RuntimeErrorBuild a deterministic conninfo string.
Compose conninfo from explicit env keys (host, dbname, user, password).
Avoid hidden defaults for critical connection fields.
Validate any interpolated SQL identifiers. Only allow safe schema/table identifier patterns (letters, digits, underscore, leading alpha/underscore). Reject invalid identifiers before building SQL.
Initialize and open AsyncConnectionPool.
Use a module-level pool and explicit startup/shutdown hooks.
Create required schema/table idempotently with CREATE ... IF NOT EXISTS.
Keep each DB helper focused.
Each function should do one clear DB task (fetch, append, close, aggregate).
Use async with pool.connection() and async with conn.transaction() for write paths.
Parameterize values, not identifiers.
Use %s bind params for data values.
Only interpolate pre-validated identifiers (schema/table names).
Keep JSONB logic explicit.
Read JSONB into simple dict/list structures.
Guard shape assumptions (dict, list, keys present).
Mutate minimal fields and write back once.
Handle time semantics consistently. Use timezone-aware timestamps. Pick one clock basis for comparisons (local-time or UTC) and apply consistently.
Return deterministic output shapes.
For aggregations, return stable, typed mappings (for example dict[str, int]).
Return empty objects instead of None where callers expect iterables/maps.
No active pool available:
Return safe default (None side-effect for writes, {} for map-return reads) unless caller contract requires hard failure.
Record exists vs does not exist: For append-style events, insert on first event; otherwise update latest row under transaction.
Missing nested JSON path:
Create missing nested containers (devices, per-device services) before append.
Incomplete/invalid historical timestamps: Skip duration calculation instead of failing the whole operation.
datetime values.Use this structural pattern for new functions:
psycopg-db-pattern to add a helper that reads latest viewrecords for one user and one service."psycopg-db-pattern to refactor this ORM query into async psycopg with parameterized SQL."psycopg-db-pattern to add schema bootstrap for a new JSONB events table."