Databases Basics

9 examples for Python database fundamentals - 6 basic and 3 intermediate.

Prerequisites

uv pip install "sqlalchemy>=2.0" alembic asyncpg psycopg[binary]

• SQLAlchemy 2.0 style is assumed throughout this section.

Basic Examples

1. DB-API Connection

Connect with a PEP 249 driver.

import sqlite3
 
with sqlite3.connect("app.db") as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS items (id INTEGER PRIMARY KEY, name TEXT)")
    conn.execute("INSERT INTO items(name) VALUES (?)", ("widget",))

• Context manager commits on success.
• Parameterized queries prevent SQL injection.
• SQLite for local dev only at scale.

Related: SQLAlchemy Core - engine layer

2. Transactions

Commit and rollback explicitly.

conn.execute("BEGIN")
try:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()
    raise

• Keep transactions short.
• One logical unit per transaction.
• Handle exceptions with rollback.

Related: Connection Management - pool scope

3. SQLAlchemy Engine

Create an engine and execute text.

from sqlalchemy import create_engine, text
engine = create_engine("sqlite:///app.db")
with engine.begin() as conn:
    conn.execute(text("SELECT 1"))

• Engine manages connections.
• begin() auto-commits.
• Use URL dialect strings consistently.

4. ORM Declarative Model

Map a class to a table.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
class Base(DeclarativeBase):
    pass
 
class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()

• Mapped[] annotates columns.
• DeclarativeBase is SQLAlchemy 2.0.
• Create tables via metadata or Alembic.

Related: SQLAlchemy ORM 2.0 - sessions

5. Session Query

Read rows with a session.

from sqlalchemy import select
from sqlalchemy.orm import Session
 
with Session(engine) as session:
    rows = session.scalars(select(Item)).all()

• Session tracks object state.
• select() replaces legacy query API.
• Context manager closes session.

6. Insert ORM Row

Persist a new entity.

with Session(engine) as session:
    session.add(Item(name="bolt"))
    session.commit()

• add() attaches to session.
• commit() flushes SQL.
• refresh() reloads generated keys.

Intermediate Examples

7. Alembic Init

Start migration history.

# alembic init alembic
# edit alembic.ini sqlalchemy.url
# alembic revision --autogenerate -m "init"
# alembic upgrade head

• Autogenerate diffs models vs DB.
• Review migrations manually.
• Upgrade in deploy pipeline.

Related: Migrations with Alembic - workflows

8. Async Engine

Non-blocking PostgreSQL access.

from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

• asyncpg driver for Postgres.
• Pair with async FastAPI routes.
• pool_pre_ping avoids stale sockets.

Related: Async Database Access - sessions

9. Redis Cache

Cache hot reads.

import redis
r = redis.Redis.from_url("redis://localhost:6379/0")
r.setex("item:1", 60, "cached-payload")

• TTL keys expire automatically.
• Cache aside pattern in services.
• Treat cache as optional - source of truth stays DB.

Related: NoSQL & Caching - patterns

Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.