Use sqlite3 and SQLAlchemy safely with parameters, explicit transactions, and deliberate abstraction.

SQL databases in Python

Relational databases provide constraints, joins, transactions, and durable shared state. Python's standard library includes SQLite support; SQLAlchemy adds composable SQL, mapping, connection pools, and portability across database engines.

Choose the smallest sufficient layer

Use sqlite3 when:

the application is local, embedded, or single-service;
SQLite's concurrency and operational model fit;
the schema and queries are small enough to keep direct SQL clear;
minimizing dependencies matters.

Use SQLAlchemy when:

the application may use PostgreSQL, MySQL, or multiple engines;
connection pooling and transaction composition matter;
a richer query builder or ORM reduces repeated mapping code;
the team benefits from a consistent data-access layer.

An ORM does not remove the need to understand SQL, indexes, query plans, isolation, and constraints.

Safe `sqlite3` setup

import sqlite3
from pathlib import Path


def connect(path: Path) -> sqlite3.Connection:
    connection = sqlite3.connect(path)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys = ON")
    return connection


SCHEMA = """
CREATE TABLE IF NOT EXISTS projects (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL UNIQUE,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY,
    project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
    title TEXT NOT NULL,
    done INTEGER NOT NULL DEFAULT 0 CHECK (done IN (0, 1))
);

CREATE INDEX IF NOT EXISTS idx_tasks_project_id ON tasks(project_id);
"""


def initialize(connection: sqlite3.Connection) -> None:
    connection.executescript(SCHEMA)

Enable foreign keys on every SQLite connection. They are not reliably enforced merely because the schema declares them.

Always parameterize values

def find_tasks(
    connection: sqlite3.Connection,
    project_id: int,
    query: str,
) -> list[sqlite3.Row]:
    cursor = connection.execute(
        """
        SELECT id, title, done
        FROM tasks
        WHERE project_id = ? AND title LIKE ?
        ORDER BY id
        """,
        (project_id, f"%{query}%"),
    )
    return cursor.fetchall()

Parameters protect values from SQL injection and preserve correct quoting. Never interpolate user input with f-strings:

# Unsafe: input becomes executable SQL syntax.
connection.execute(f"SELECT * FROM tasks WHERE title = '{title}'")

Placeholders cannot parameterize SQL identifiers such as table names or sort directions. If those must be dynamic, map an external enum to a hard-coded allowlist.

SORT_COLUMNS = {"title": "title", "created": "id"}


def order_column(requested: str) -> str:
    try:
        return SORT_COLUMNS[requested]
    except KeyError as exc:
        raise ValueError("unsupported sort column") from exc

Make transaction boundaries visible

def create_project_with_tasks(
    connection: sqlite3.Connection,
    name: str,
    titles: list[str],
) -> int:
    with connection:
        cursor = connection.execute(
            "INSERT INTO projects (name) VALUES (?)",
            (name,),
        )
        project_id = cursor.lastrowid
        if project_id is None:
            raise RuntimeError("database did not return a project id")

        connection.executemany(
            "INSERT INTO tasks (project_id, title) VALUES (?, ?)",
            ((project_id, title) for title in titles),
        )
    return project_id

The connection context manager commits on success and rolls back when the block raises. It does not close the connection. Keep one business operation inside one transaction; do not commit after each row.

For new code, consider explicit transaction mode:

connection = sqlite3.connect(path, autocommit=False)

Python version and legacy behavior matter, so test transaction semantics on every supported Python version. Do not mix manual BEGIN statements with context-manager behavior unless you understand the driver mode.

SQLAlchemy 2.x Core

uv add "sqlalchemy>=2,<3"

from sqlalchemy import (
    Boolean,
    Column,
    ForeignKey,
    Integer,
    MetaData,
    String,
    Table,
    create_engine,
    select,
)

metadata = MetaData()

projects = Table(
    "projects",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(120), nullable=False, unique=True),
)

tasks = Table(
    "tasks",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("project_id", ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
    Column("title", String(240), nullable=False),
    Column("done", Boolean, nullable=False, default=False),
)

engine = create_engine("sqlite+pysqlite:///tasks.db")
metadata.create_all(engine)


def task_titles(project_id: int) -> list[str]:
    statement = (
        select(tasks.c.title)
        .where(tasks.c.project_id == project_id)
        .order_by(tasks.c.id)
    )
    with engine.connect() as connection:
        return list(connection.scalars(statement))

SQLAlchemy binds Python values as parameters. Use text() plus named parameters when direct SQL is clearer:

from sqlalchemy import text

statement = text(
    "SELECT id, title FROM tasks WHERE project_id = :project_id ORDER BY id"
)

with engine.connect() as connection:
    rows = connection.execute(statement, {"project_id": 7}).mappings().all()

SQLAlchemy ORM with typed mappings

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class Project(Base):
    __tablename__ = "projects"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120), unique=True)
    tasks: Mapped[list["Task"]] = relationship(
        back_populates="project",
        cascade="all, delete-orphan",
    )


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    project_id: Mapped[int] = mapped_column(ForeignKey("projects.id"))
    title: Mapped[str] = mapped_column(String(240))
    done: Mapped[bool] = mapped_column(default=False)
    project: Mapped[Project] = relationship(back_populates="tasks")

Use one session per request or unit of work:

from sqlalchemy.orm import Session


def create_project(name: str) -> int:
    with Session(engine) as session, session.begin():
        project = Project(name=name)
        session.add(project)
        session.flush()
        return project.id

flush() sends pending SQL so generated IDs become available; the transaction still commits only when the context exits successfully.

Schema migrations

metadata.create_all() creates missing tables; it does not safely evolve existing production schemas. Use a migration tool such as Alembic:

uv add alembic
uv run alembic init migrations
uv run alembic revision --autogenerate -m "add task due date"
uv run alembic upgrade head

Review autogenerated migrations. Renames can look like drop-and-create operations, and data backfills require deliberate ordering.

Common pitfalls

Sharing one SQLite connection across threads without a clear concurrency design.
Forgetting PRAGMA foreign_keys = ON.
Building SQL from user input, especially ORDER BY fragments.
Keeping transactions open while waiting for network calls.
Catching integrity errors without rolling back the failed SQLAlchemy session.
Returning lazy ORM relationships after the session has closed.
Assuming SQLite's typing and concurrency match production PostgreSQL.
Using create_all() as a migration strategy.

Practice

Create projects and tasks with foreign-key and uniqueness constraints.
Write a transaction test proving a failed second insert rolls back the first.
Add a safe, allowlisted sort option.
Implement the same query with sqlite3 and SQLAlchemy Core; compare clarity.
Inspect a query plan and add an index justified by the observed lookup pattern.

SQL databases in Python

SQL databases in Python

Choose the smallest sufficient layer

Safe `sqlite3` setup

Always parameterize values

Make transaction boundaries visible

SQLAlchemy 2.x Core

SQLAlchemy ORM with typed mappings

Schema migrations

Common pitfalls

Practice

Finish this lesson

SQL databases in Python

On this page