Python PostgreSQL Tutorial: Learn Database from Scratch (2026)
PostgreSQL is my default choice for any application that needs a relational database — it's battle-tested, has excellent standards compliance, and supports advanced features like JSONB, full-text search, and window functions. The psycopg2 library is the de facto Python adapter, known for its stability and support for server-side cursors and COPY operations for bulk data loading.
This tutorial covers the full PostgreSQL + Python workflow: connecting, executing queries, using connection pools, leveraging PostgreSQL-specific features like JSONB and RETURNING, and integrating with SQLAlchemy for ORM-based access.
Connecting to PostgreSQL with psycopg2
psycopg2.connect() returns a connection object. The connection string can be a DSN or individual parameters. Use the with statement to ensure the connection is properly closed. The cursor object executes queries and fetches results. psycopg2's RealDictCursor returns rows as dicts instead of tuples, which is more readable.
import psycopg2
from psycopg2 import extras
try:
conn = psycopg2.connect(
host='localhost',
database='blog',
user='postgres',
password='your_password',
port=5432
)
cursor = conn.cursor(cursor_factory=extras.RealDictCursor)
cursor.execute("SELECT version()")
print(cursor.fetchone()['version'])
except psycopg2.Error as e:
print(f"Error: {e}")
finally:
if conn:
conn.close()Creating Tables with PostgreSQL Data Types
PostgreSQL supports a rich set of data types: SERIAL for auto-increment, NUMERIC for exact decimal arithmetic, JSONB for JSON storage with indexing, ARRAY, TSVECTOR for full-text search, and UUID. The SERIAL type auto-increments. Use IF NOT EXISTS to make migrations idempotent.
create_users = """
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(255) NOT NULL,
profile JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW()
)"""
create_posts = """
CREATE TABLE IF NOT EXISTS posts (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
title VARCHAR(200) NOT NULL,
body TEXT,
tags TEXT[],
search_vector TSVECTOR,
published BOOLEAN DEFAULT false,
created_at TIMESTAMPTZ DEFAULT NOW()
)"""
cursor.execute(create_users)
cursor.execute(create_posts)
conn.commit()INSERT, UPDATE, DELETE with RETURNING
PostgreSQL's RETURNING clause returns values from the affected row — useful for getting auto-generated IDs and defaults without a separate query. Use the %s placeholders for parameterized queries to prevent SQL injection. executemany() inserts multiple rows efficiently.
# INSERT with RETURNING
cursor.execute(
"""INSERT INTO users (username, email, profile)
VALUES (%s, %s, %s)
RETURNING id, created_at""",
('alice', 'alice@example.com', '{"theme": "dark"}')
)
user_id, created_at = cursor.fetchone()['id'], cursor.fetchone()['created_at']
print(f"Created user {user_id} at {created_at}")
conn.commit()
# UPDATE with RETURNING
cursor.execute(
"UPDATE users SET email = %s WHERE id = %s RETURNING username",
('newemail@example.com', user_id)
)
print(f"Updated: {cursor.fetchone()['username']}")
conn.commit()
# DELETE with RETURNING
cursor.execute(
"DELETE FROM users WHERE id = %s RETURNING username",
(999,)
)
if cursor.rowcount == 0:
print("User not found")Advanced Queries: JSONB, Full-Text Search, and CTEs
PostgreSQL's JSONB operators (->, ->>, @>, ?) let you query JSON data inside relational tables. Full-text search uses tsvector and tsquery with the @@ operator. Common Table Expressions (WITH queries) enable recursive queries and modular SQL. These features reduce the need for application-level processing.
# JSONB query: users with dark theme
dark_theme_users = cursor.execute(
"SELECT username FROM users WHERE profile @> '{"theme": "dark"}'::jsonb"
)
# Full-text search
cursor.execute("""
SELECT title, ts_rank(search_vector, plainto_tsquery('english', %s)) AS rank
FROM posts
WHERE search_vector @@ plainto_tsquery('english', %s)
ORDER BY rank DESC
LIMIT 10
""", ('database tutorial', 'database tutorial'))
# CTE for hierarchical data
cursor.execute("""
WITH RECURSIVE comment_tree AS (
SELECT id, parent_id, body, 1 AS depth
FROM comments WHERE parent_id IS NULL
UNION ALL
SELECT c.id, c.parent_id, c.body, ct.depth + 1
FROM comments c
JOIN comment_tree ct ON c.parent_id = ct.id
)
SELECT * FROM comment_tree ORDER BY depth, id
""")Connection Pooling with psycopg2.pool
ThreadedConnectionPool manages a set of persistent connections that are reused across requests. This avoids the overhead of establishing a TCP connection for every operation. getconn() acquires a connection, putconn() returns it. For async applications, use asyncpg instead of psycopg2.
from psycopg2.pool import ThreadedConnectionPool
pool = ThreadedConnectionPool(
minconn=2,
maxconn=10,
host='localhost',
database='blog',
user='postgres',
password='your_password'
)
def get_user(user_id):
conn = pool.getconn()
try:
with conn.cursor(cursor_factory=extras.RealDictCursor) as cur:
cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
return cur.fetchone()
finally:
pool.putconn(conn)
user = get_user(1)
print(user)
pool.closeall()Using PostgreSQL with SQLAlchemy ORM
SQLAlchemy's ORM maps Python classes to PostgreSQL tables. The declarative base system defines models. Relationship() handles foreign key joins automatically. SQLAlchemy also supports PostgreSQL-specific types like JSONB, ARRAY, and TSVECTOR through its PostgreSQL dialect.
from sqlalchemy import create_engine, Column, Integer, String, Text, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.dialects.postgresql import JSONB, ARRAY
engine = create_engine('postgresql://user:pass@localhost/blog')
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(255), nullable=False)
profile = Column(JSONB, default={})
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
new_user = User(username='bob', email='bob@example.com', profile={'theme': 'light'})
session.add(new_user)
session.commit()
users = session.query(User).filter(User.profile['theme'].astext == 'light').all()
print(f"Light theme users: {len(users)}")Frequently Asked Questions
Should I use psycopg2 or asyncpg?
psycopg2 is synchronous and blocks the event loop. Use asyncpg for async applications (FastAPI, aiohttp). asyncpg is also faster for raw queries. psycopg2 has broader ecosystem support and is more stable for traditional web apps.
How do I handle PostgreSQL connection timeouts?
Set connect_timeout in the connection string (seconds). For long-running queries, set statement_timeout at the session level. Use keepalives to detect dead connections: keepalives=1, keepalives_idle=30, keepalives_interval=10.
What is the advantage of JSONB over a separate table?
JSONB stores flexible, schema-less data in a column with indexing support. Use it for attributes that vary between rows or change frequently. Use a separate table when you need referential integrity or query individual values as part of complex joins.
How do I backup a PostgreSQL database from Python?
Use subprocess to call pg_dump or pg_dumpall. The psycopg2 COPY operations (copy_to, copy_from) are useful for table-level export/import. For production, manage backups at the database level, not from application code.
Originally published on Ayodhyyya. Last updated June 1, 2026.