🐍 Python and PostgreSQL Integration

Three Levels of Database Access

Python can talk to PostgreSQL at three different heights of abstraction. Each trades control for convenience.

💡 An analogy — levels of driving assistance: A raw driver (Level 1) is a manual transmission: total control, every gear change is yours. A query helper (Level 2) is an automatic with cruise control — you still steer, but the tedious bits are handled. An ORM (Level 3) is adaptive cruise with lane-keeping: you set the destination in Python objects and it writes the SQL. None is "best" — different roads call for different amounts of assistance.

This lesson lives at Level 1. Understanding the driver makes the ORM in the next lesson feel like magic you can explain, not magic you fear.

Connecting Safely

First install the driver into a virtual environment:

python -m venv venv
source venv/bin/activate        # Windows: venv\Scripts\activate

pip install "psycopg[binary]"   # psycopg3 with prebuilt binaries
# (older projects: pip install psycopg2-binary)

The safest connection pattern uses a with block, which guarantees the connection is committed-or-rolled-back and closed even if an error is raised:

import os
import psycopg

# Build a connection string from environment variables
conninfo = (
    f"host={os.environ.get('DB_HOST', 'localhost')} "
    f"port={os.environ.get('DB_PORT', '5432')} "
    f"dbname={os.environ['DB_NAME']} "
    f"user={os.environ['DB_USER']} "
    f"password={os.environ['DB_PASSWORD']}"
)

# The 'with' block manages the transaction and closes the connection
with psycopg.connect(conninfo) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT version();")
        print(cur.fetchone()[0])
# On a clean exit the transaction is committed; on an exception it is rolled back.

url = os.environ["DATABASE_URL"]   # postgresql://user:pass@host:5432/dbname
with psycopg.connect(url) as conn:
    ...

Parameterized Queries

This is the single most important habit in this lesson. Never build SQL by pasting user input into a string. Doing so opens the door to SQL injection, where a crafted input rewrites your query.

# DANGER — never do this
username = request_input   # imagine: "'; DROP TABLE users; --"
cur.execute(f"SELECT * FROM users WHERE username = '{username}'")
# The attacker's text becomes part of your SQL.

Instead, pass values separately using %s placeholders. The driver sends the query and the data apart, so the data can never be interpreted as SQL:

# SAFE — the driver escapes the value for you
username = "johndoe"
cur.execute("SELECT id, email FROM users WHERE username = %s", (username,))
user = cur.fetchone()
print(user)   # e.g. (1, 'john@example.com')

RETURNING gives you generated values

When you insert a row with an identity key, ask PostgreSQL to hand the new id straight back:

import datetime

cur.execute(
    """
    INSERT INTO orders (customer_id, items, total, ordered_at, is_shipped, notes)
    VALUES (%s, %s, %s, %s, %s, %s)
    RETURNING id
    """,
    (
        42,                                   # integer
        ["Widget", "Gadget"],                 # -> a PostgreSQL text[] array
        99.95,                                # numeric
        datetime.datetime.now(datetime.UTC),  # timestamptz
        True,                                 # boolean
        "Priority shipping",                  # text
    ),
)
new_id = cur.fetchone()[0]
print(f"Created order {new_id}")

Notice psycopg maps Python types to PostgreSQL types for free: a list becomes an array, a datetime becomes a timestamp, a dict can become JSON.

Transactions

A transaction groups several statements so they all succeed or all fail together — the "A" (atomicity) in ACID. The classic example is a bank transfer: debiting one account and crediting another must never happen halfway.

With psycopg3, the with conn: block is the transaction boundary. Exiting cleanly commits; raising an exception rolls back:

with psycopg.connect(conninfo) as conn:
    with conn.cursor() as cur:
        cur.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (100, 1),
        )
        cur.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (100, 2),
        )
    # Reaching here without an exception commits both updates atomically.
    # If either statement raised, neither change is saved.

Deciding to roll back yourself

Sometimes the data itself tells you to abort — for example, insufficient funds. Raise an exception to trigger the automatic rollback:

class InsufficientFunds(Exception):
    pass

try:
    with psycopg.connect(conninfo) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT balance FROM accounts WHERE id = %s FOR UPDATE", (1,))
            balance = cur.fetchone()[0]
            if balance < 100:
                raise InsufficientFunds("Not enough money")
            cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
            cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
except InsufficientFunds as e:
    print(f"Transfer cancelled: {e}")   # the whole transaction was rolled back

Cursors & Fetching Results

A cursor runs a query and steps through its results. By default rows come back as plain tuples, addressed by position:

with conn.cursor() as cur:
    cur.execute("SELECT id, username, email FROM users")
    row = cur.fetchone()          # one row, or None
    some = cur.fetchmany(3)       # a list of up to 3 rows
    rest = cur.fetchall()         # every remaining row
    # Cursors are iterable too:
    for r in cur:
        print(r[0], r[1])

Getting rows as dictionaries

Accessing row[2] is fragile — reorder the columns and your code breaks. Ask for dictionary rows and address fields by name:

from psycopg.rows import dict_row

with conn.cursor(row_factory=dict_row) as cur:
    cur.execute("SELECT id, username, email FROM users LIMIT 1")
    user = cur.fetchone()
    print(user["username"], user["email"])   # readable and reorder-proof

psycopg2 equivalent: pass cursor_factory=psycopg2.extras.RealDictCursor instead.

Huge result sets: server-side cursors

Calling fetchall() on a million-row table loads a million rows into memory. A named (server-side) cursor streams them in batches instead:

with conn.cursor(name="big_export") as cur:   # a name makes it server-side
    cur.execute("SELECT * FROM logs")
    while batch := cur.fetchmany(1000):
        for row in batch:
            process(row)                      # memory stays flat

Bulk Operations & Pooling

Loading many rows fast with COPY

Inserting rows one execute at a time is fine for a handful, painfully slow for thousands. PostgreSQL's COPY is the fastest path for bulk loads, and psycopg3 exposes it cleanly:

rows = [
    ("INFO", f"Log message {i}") for i in range(1, 100_001)
]

with conn.cursor() as cur:
    with cur.copy("COPY logs (level, message) FROM STDIN") as copy:
        for row in rows:
            copy.write_row(row)
# 100,000 rows in a fraction of the time a loop of INSERTs would take.

For a moderate batch where you still want plain SQL, executemany is a good middle ground:

cur.executemany(
    "INSERT INTO logs (level, message) VALUES (%s, %s)",
    rows[:1000],
)

Connection pooling

Opening a new connection per request is expensive — the TCP handshake and authentication add up. A connection pool keeps a set of connections warm and hands them out on demand. In psycopg3 the pool lives in a separate package:

pip install "psycopg[pool]"

from psycopg_pool import ConnectionPool

# Create one pool for the whole application (min 2, max 10 connections)
pool = ConnectionPool(conninfo, min_size=2, max_size=10)

def get_user(user_id: int):
    # Borrow a connection; it returns to the pool automatically at block exit
    with pool.connection() as conn:
        with conn.cursor(row_factory=dict_row) as cur:
            cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            return cur.fetchone()

# On application shutdown:
pool.close()

Error Handling

psycopg raises a hierarchy of exceptions that mirror PostgreSQL's error conditions. Catching specific ones lets you respond intelligently — a duplicate key is a user-facing "already taken", a connection drop is a retry.

import psycopg
from psycopg import errors

try:
    with psycopg.connect(conninfo) as conn:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO users (username, email) VALUES (%s, %s)",
                ("johndoe", "john@example.com"),
            )
except errors.UniqueViolation:
    print("That username or email is already taken.")   # a 409 to the user
except psycopg.OperationalError as e:
    print(f"Database unavailable, will retry: {e}")      # transient — retry
except psycopg.Error as e:
    print(f"Unexpected database error: {e}")             # catch-all fallback

Hands-on Exercise

CREATE TABLE accounts (
    id       INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    owner    TEXT NOT NULL,
    balance  NUMERIC(12,2) NOT NULL CHECK (balance >= 0)
);
INSERT INTO accounts (owner, balance) VALUES ('Ana', 500), ('Ben', 100);

from decimal import Decimal
from psycopg_pool import ConnectionPool

class InsufficientFunds(Exception):
    pass

def transfer(pool: ConnectionPool, from_id: int, to_id: int, amount: Decimal) -> None:
    if amount <= 0:
        raise ValueError("amount must be positive")
    with pool.connection() as conn:            # this block is the transaction
        with conn.cursor() as cur:
            cur.execute(
                "SELECT balance FROM accounts WHERE id = %s FOR UPDATE",
                (from_id,),
            )
            row = cur.fetchone()
            if row is None:
                raise ValueError("source account not found")
            if row[0] < amount:
                raise InsufficientFunds(f"balance {row[0]} < {amount}")

            cur.execute(
                "UPDATE accounts SET balance = balance - %s WHERE id = %s",
                (amount, from_id),
            )
            cur.execute(
                "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                (amount, to_id),
            )
        # clean exit -> COMMIT; any raise above -> ROLLBACK

# Usage
pool = ConnectionPool(conninfo, min_size=1, max_size=5)
try:
    transfer(pool, 1, 2, Decimal("150.00"))
    print("Transfer complete")
except InsufficientFunds as e:
    print(f"Declined: {e}")

Summary & Quiz

📚 Further Reading

• psycopg 3 Documentation
• psycopg — Connection Pools
• PostgreSQL — COPY

🚀 What's Next?

Writing raw SQL by hand is powerful but verbose. Next, ORM Basics with SQLAlchemy shows how to map Python classes to tables so you can work with objects instead of strings — while still dropping down to SQL whenever you need to.