Typed PostgreSQL Access from Python and TypeScript

Application database access should make query inputs, output shape, transaction ownership, tenant scope, timeouts, and error behavior explicit. Python’s psycopg and TypeScript’s node-postgres both support server parameters and pooling; neither automatically provides authorization, schema compatibility, idempotency, safe retries, or correct transaction boundaries. Keep domain operations narrow and observable.

What will you be able to do?

Execute parameterized queries in Python and TypeScript.
Manage transactions with guaranteed release/rollback.
Map database rows into validated application types.
Configure pool budgets, timeouts, errors, and traces.

Prerequisites and mental model

Use four layers: route/handler authenticates; application service owns the use case/transaction; repository or query module executes known SQL; PostgreSQL enforces durable invariants. Avoid an abstraction that hides SQL semantics and avoid SQL scattered through controllers.

What does a Python operation look like?

python

from dataclasses import dataclass
from psycopg_pool import ConnectionPool

@dataclass(frozen=True)
class TicketSummary:
    id: int
    subject: str
    status: str

def get_ticket(pool: ConnectionPool, organization_id: int, ticket_id: int) -> TicketSummary | None:
    with pool.connection() as conn:
        with conn.transaction():
            conn.execute("SELECT set_config('app.organization_id', %s, true)",
                         (str(organization_id),))
            row = conn.execute(
                """SELECT id, subject, status
                   FROM signaldesk.tickets
                   WHERE organization_id = %s AND id = %s""",
                (organization_id, ticket_id),
            ).fetchone()
    return None if row is None else TicketSummary(*row)

What does TypeScript look like?

const client = await pool.connect();
try {
  await client.query("BEGIN");
  await client.query(
    "SELECT set_config('app.organization_id', $1, true)",
    [organizationId],
  );
  const result = await client.query(
    `SELECT id, subject, status
     FROM signaldesk.tickets
     WHERE organization_id = $1 AND id = $2`,
    [organizationId, ticketId],
  );
  await client.query("COMMIT");
  return result.rows[0] ?? null;
} catch (error) {
  await client.query("ROLLBACK");
  throw error;
} finally {
  client.release();
}

Production code validates returned row shape, configures connect/statement/lock timeouts, maps SQLSTATEs to safe domain errors, and never returns raw database messages to untrusted clients.

Why does this matter in AI-native systems?

Release database connections before model calls. Validate model output into typed commands, reacquire a connection, set tenant context, recheck state/version, and commit briefly. Trace query fingerprints and timings without logging customer data or vectors.

Prove it worked

Integration tests cover success, missing row, other tenant, constraint violation, timeout, pool exhaustion, rollback, and dropped connection. Assert the pool returns to baseline and no tenant setting leaks between transactions.

Production judgment

Driver defaults and APIs change; pin and read current docs.
ORMs do not remove SQL planning, transaction, migration, or N+1 concerns.
Retry only classified transient failures at idempotent transaction boundaries.
Cancellation and request timeouts should propagate to database work.

Common mistakes

Connection leak: release omitted on error → structured context/finally.
SQL injection despite driver: dynamic identifier concatenation → fixed allowlist/composition.
Transaction unexpectedly commits: driver context semantics misunderstood → test failure path.

AI Pair-Programmer Prompt

text

Review this application data-access operation. Check authentication versus authorization,
tenant transaction context, SQL parameters, typed input/output, transaction owner,
connection release, timeouts/cancellation, SQLSTATE mapping, retries/idempotency,
observability/redaction, model-call boundaries, and integration tests. Do not hide SQL.

Hands-on exercise

Implement resolve_ticket(expected_version) in one language with a conditional update and audit event.

Acceptance criterion: success is atomic, stale/other-tenant cases change nothing, connections release on all paths, and model calls are outside the transaction.

Knowledge check

What do parameters not provide?
Where should transaction ownership live?
Why validate returned rows?

Answers

Authorization, correct semantics, limits, or safe retries.
The application use case/service that knows the invariant.
Schema/drivers return runtime data that must match the application contract.

Completion checklist

Queries are known and parameterized.
Transactions and connections close on every path.
Tenant context is transaction-local.
Negative and saturation integration tests pass.

Primary references

Typed PostgreSQL Access from Python and TypeScript

What does a Python operation look like?

python

from dataclasses import dataclass
from psycopg_pool import ConnectionPool

@dataclass(frozen=True)
class TicketSummary:
    id: int
    subject: str
    status: str

def get_ticket(pool: ConnectionPool, organization_id: int, ticket_id: int) -> TicketSummary | None:
    with pool.connection() as conn:
        with conn.transaction():
            conn.execute("SELECT set_config('app.organization_id', %s, true)",
                         (str(organization_id),))
            row = conn.execute(
                """SELECT id, subject, status
                   FROM signaldesk.tickets
                   WHERE organization_id = %s AND id = %s""",
                (organization_id, ticket_id),
            ).fetchone()
    return None if row is None else TicketSummary(*row)

What does TypeScript look like?

const client = await pool.connect();
try {
  await client.query("BEGIN");
  await client.query(
    "SELECT set_config('app.organization_id', $1, true)",
    [organizationId],
  );
  const result = await client.query(
    `SELECT id, subject, status
     FROM signaldesk.tickets
     WHERE organization_id = $1 AND id = $2`,
    [organizationId, ticketId],
  );
  await client.query("COMMIT");
  return result.rows[0] ?? null;
} catch (error) {
  await client.query("ROLLBACK");
  throw error;
} finally {
  client.release();
}

Production code validates returned row shape, configures connect/statement/lock timeouts, maps SQLSTATEs to safe domain errors, and never returns raw database messages to untrusted clients.

AI Pair-Programmer Prompt

text

Review this application data-access operation. Check authentication versus authorization,
tenant transaction context, SQL parameters, typed input/output, transaction owner,
connection release, timeouts/cancellation, SQLSTATE mapping, retries/idempotency,
observability/redaction, model-call boundaries, and integration tests. Do not hide SQL.

Knowledge check

What do parameters not provide?

Where should transaction ownership live?

Why validate returned rows?

Answers

Authorization, correct semantics, limits, or safe retries.

The application use case/service that knows the invariant.

Schema/drivers return runtime data that must match the application contract.