Working with PostgreSQL from Python

psycopg is the standard PostgreSQL driver for Python. Connect with a connection string (keep it in the DATABASE_URL environment variable) and run SQL with conn.execute.

Always pass values as parameters (%s placeholders plus a tuple) — never build SQL with f-strings. Parameters are sent separately from the SQL, which completely prevents SQL injection.

with conn.transaction(): groups statements so they all succeed or all roll back. In web apps, use a connection pool (psycopg_pool) instead of opening a new connection for every request. See the PostgreSQL track on this site to learn the SQL itself.

TerminalShell

pip install "psycopg[binary]" psycopg_pool
export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/school"

db.pyPython

import os
from dataclasses import dataclass

import psycopg
from psycopg.rows import class_row

DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/school")


@dataclass
class Result:
    name: str
    subject: str
    score: int


def main() -> None:
    with psycopg.connect(DATABASE_URL) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS py_results (
                id      bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                name    text NOT NULL,
                subject text NOT NULL,
                score   int  NOT NULL CHECK (score BETWEEN 0 AND 100)
            )
        """)
        conn.execute("TRUNCATE py_results")

        rows = [("Amina", "Maths", 88), ("Juma", "Maths", 42), ("Neema", "Biology", 71)]
        with conn.transaction():                     # all-or-nothing
            with conn.cursor() as cur:
                cur.executemany(
                    "INSERT INTO py_results (name, subject, score) VALUES (%s, %s, %s)", rows
                )

        name = "Amina'; DROP TABLE py_results; --"   # malicious input is harmless as a parameter
        print(conn.execute("SELECT count(*) FROM py_results WHERE name = %s", (name,)).fetchone())

        with conn.cursor(row_factory=class_row(Result)) as cur:
            cur.execute(
                "SELECT name, subject, score FROM py_results WHERE score >= %s ORDER BY score DESC",
                (45,),
            )
            for r in cur.fetchall():
                print(r)

        try:
            with conn.transaction():
                conn.execute("INSERT INTO py_results (name, subject, score) VALUES ('Ali', 'Maths', 50)")
                conn.execute("INSERT INTO py_results (name, subject, score) VALUES ('Bad', 'Maths', 150)")
        except psycopg.errors.CheckViolation:
            print("rolled back; Ali was not saved either")

        print(conn.execute("SELECT count(*) FROM py_results").fetchone())   # (3,)


if __name__ == "__main__":
    main()

Key points

Never format SQL with f-strings — always use %s parameters.
conn.transaction() makes a group of statements all-or-nothing.
Use a connection pool in servers; keep the connection string in an environment variable.

Exercise

Connect the FastAPI app from the previous lesson to PostgreSQL: create a students table, use a psycopg_pool.ConnectionPool opened at startup, and replace the in-memory store with SQL queries.

Show solution

Try the exercise yourself first — then compare your approach with this one.

A ConnectionPool is opened once when the app starts (in the lifespan handler) and closed on shutdown. Each route borrows a connection with pool.connection(), runs parameterised SQL, and returns rows mapped to Pydantic models with class_row.

main.pyPython

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, status
from psycopg.rows import class_row
from psycopg_pool import ConnectionPool
from pydantic import BaseModel, Field

DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/school")
pool = ConnectionPool(DATABASE_URL, min_size=1, max_size=5, open=False)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    pool.open()
    with pool.connection() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS api_students (
                id   bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                name text     NOT NULL,
                form smallint NOT NULL CHECK (form BETWEEN 1 AND 6)
            )
        """)
    yield
    pool.close()


app = FastAPI(lifespan=lifespan)


class StudentIn(BaseModel):
    name: str = Field(min_length=2)
    form: int = Field(ge=1, le=6)


class StudentOut(BaseModel):
    id: int
    name: str
    form: int


@app.post("/students", status_code=status.HTTP_201_CREATED)
def create_student(body: StudentIn) -> StudentOut:
    with pool.connection() as conn, conn.cursor(row_factory=class_row(StudentOut)) as cur:
        cur.execute(
            "INSERT INTO api_students (name, form) VALUES (%s, %s) RETURNING id, name, form",
            (body.name, body.form),
        )
        return cur.fetchone()


@app.get("/students/{student_id}")
def get_student(student_id: int) -> StudentOut:
    with pool.connection() as conn, conn.cursor(row_factory=class_row(StudentOut)) as cur:
        cur.execute("SELECT id, name, form FROM api_students WHERE id = %s", (student_id,))
        student = cur.fetchone()
    if student is None:
        raise HTTPException(status_code=404, detail="Student not found")
    return student


@app.get("/students")
def list_students(form: int | None = None) -> list[StudentOut]:
    with pool.connection() as conn, conn.cursor(row_factory=class_row(StudentOut)) as cur:
        cur.execute(
            "SELECT id, name, form FROM api_students WHERE %(form)s::int IS NULL OR form = %(form)s ORDER BY id",
            {"form": form},
        )
        return cur.fetchall()

Key points

Exercise

Check your understanding