Connecting Python to PostgreSQL

PostgreSQL is a powerful open-source database known for correctness, rich data types (JSONB, arrays, ranges) and advanced SQL. The modern Python driver is psycopg 3 (package psycopg); its predecessor psycopg2 is still found in many existing projects and has a very similar API.

This lesson shows how to connect, run queries with parameters, get rows as dictionaries, use RETURNING, manage transactions, handle specific PostgreSQL errors, and use a connection pool.

These examples need a running PostgreSQL server, so run them on your own computer.

Setup

Run PostgreSQL locally or with Docker: docker run -e POSTGRES_PASSWORD=secret -p 5432:5432 postgres:17.
Install the driver: pip install "psycopg[binary]", and pip install psycopg-pool for pooling.
Connect with a connection string (DSN) such as postgresql://user:password@localhost:5432/dbname, usually read from a DATABASE_URL environment variable.

psycopg 3 Essentials

with psycopg.connect(dsn) as conn: commits when the block succeeds, rolls back on an exception, and closes the connection. conn.execute() is a shortcut that creates a cursor. Placeholders are %s or %(name)s. Pass row_factory=dict_row to receive dictionaries. INSERT ... RETURNING id returns generated values in the same round-trip. with conn.transaction(): creates an explicit transaction block (or a savepoint when nested). Specific errors live in psycopg.errors, for example UniqueViolation and ForeignKeyViolation.

Examples

Connect, create, insert with RETURNING, and query as dictionaries

Python

# pip install "psycopg[binary]"
import os
import psycopg
from psycopg.rows import dict_row

DSN = os.environ.get("DATABASE_URL", "postgresql://shop_app:secret@localhost:5432/shop")

with psycopg.connect(DSN, row_factory=dict_row) as conn:      # commits and closes at the end
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                title TEXT NOT NULL,
                done BOOLEAN NOT NULL DEFAULT false,
                created_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )
        """)
        cur.execute("INSERT INTO tasks (title) VALUES (%s) RETURNING id", ("Write report",))
        print("new id:", cur.fetchone()["id"])

        cur.executemany("INSERT INTO tasks (title) VALUES (%s)", [("Review PR",), ("Deploy",)])
        cur.execute("UPDATE tasks SET done = true WHERE title = %(title)s", {"title": "Review PR"})

        cur.execute("SELECT id, title, done FROM tasks ORDER BY id")
        for row in cur:
            print(row)

Output

new id: 1
{'id': 1, 'title': 'Write report', 'done': False}
{'id': 2, 'title': 'Review PR', 'done': True}
{'id': 3, 'title': 'Deploy', 'done': False}

Transactions and catching a UniqueViolation

Python

import os
import psycopg
from psycopg import errors

DSN = os.environ["DATABASE_URL"]

with psycopg.connect(DSN) as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS users (email TEXT PRIMARY KEY, name TEXT NOT NULL)")
    try:
        with conn.transaction():                     # both inserts or neither
            conn.execute("INSERT INTO users VALUES (%s, %s)", ("asha@example.com", "Asha"))
            conn.execute("INSERT INTO users VALUES (%s, %s)", ("asha@example.com", "Asha again"))
    except errors.UniqueViolation as e:
        print("rolled back:", e.diag.message_primary)

    print("users stored:", conn.execute("SELECT count(*) FROM users").fetchone()[0])

Output

rolled back: duplicate key value violates unique constraint "users_pkey"
users stored: 0

PostgreSQL types: JSONB and arrays map to Python dicts and lists

Python

import os
import psycopg
from psycopg.types.json import Jsonb

with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
    conn.execute("""CREATE TEMP TABLE products (
        name TEXT, tags TEXT[], specs JSONB)""")
    conn.execute("INSERT INTO products VALUES (%s, %s, %s)",
                 ("Laptop", ["electronics", "office"], Jsonb({"ram_gb": 16, "ssd_gb": 512})))
    name, tags, specs = conn.execute(
        "SELECT name, tags, specs FROM products WHERE %s = ANY(tags)", ("office",)).fetchone()
    print(name, tags, specs["ram_gb"], type(specs).__name__)

Output

Laptop ['electronics', 'office'] 16 dict

A connection pool for web applications

Python

# pip install psycopg-pool
import os
from psycopg_pool import ConnectionPool

pool = ConnectionPool(os.environ["DATABASE_URL"], min_size=2, max_size=10, open=True)

def count_open_tasks():
    with pool.connection() as conn:          # borrowed from the pool, returned afterwards
        return conn.execute("SELECT count(*) FROM tasks WHERE NOT done").fetchone()[0]

print("open tasks:", count_open_tasks())
pool.close()

Output

open tasks: 2

Common Mistakes

Mixing up psycopg (version 3) and psycopg2 examples — imports and some APIs differ.
Using ? placeholders; psycopg expects %s or %(name)s.
Catching a database error and continuing to use the connection without rolling back (PostgreSQL then rejects every command in the failed transaction).
Passing a Python dict for a JSONB column without wrapping it in Jsonb().
Creating a new connection for every web request instead of using psycopg_pool.

Key Points to Remember

psycopg 3 is the modern PostgreSQL driver; with psycopg.connect(...) commits and closes.
Use %s / %(name)s placeholders, and RETURNING to get generated ids.
row_factory=dict_row gives dictionaries; arrays become lists and JSONB becomes dicts.
conn.transaction() groups statements; psycopg.errors has specific exception classes.
Use a ConnectionPool in long-running applications.

Practice the examples

Change an input, predict the result, then compare it with the output. Explain why the result changes.

Use your local project environment for these examples. Codelab currently runs Python and HTML/CSS/JavaScript; framework examples may need project dependencies.