In the previous tutorial, we learned about HTTP requests and APIs. Now let’s learn about databases — how to store, read, update, and delete data using Python.

We will cover two approaches: Python’s built-in sqlite3 module (no installation needed) and SQLAlchemy (the most popular Python ORM). By the end of this tutorial, you will know how to build data-driven applications.

What is SQLite?

SQLite is a database that stores everything in a single file. It comes built into Python — no installation, no server, no configuration. It is perfect for:

  • Small to medium applications
  • Local data storage
  • Prototyping before switching to PostgreSQL or MySQL
  • Testing

Part 1: sqlite3 — The Built-in Module

Creating a Database and Table

import sqlite3

# Connect to a database (creates the file if it doesn't exist)
conn = sqlite3.connect("tasks.db")

# Use :memory: for a temporary database (great for testing)
# conn = sqlite3.connect(":memory:")

# Enable accessing columns by name
conn.row_factory = sqlite3.Row

# Create a table
conn.execute("""
    CREATE TABLE IF NOT EXISTS tasks (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        done INTEGER DEFAULT 0,
        created_at TEXT DEFAULT CURRENT_TIMESTAMP
    )
""")
conn.commit()

conn.row_factory = sqlite3.Row lets you access columns by name (row["title"]) instead of index (row[0]). Always use this.

Insert Data

def insert_task(conn: sqlite3.Connection, title: str) -> int:
    """Insert a task and return its ID."""
    cursor = conn.execute(
        "INSERT INTO tasks (title) VALUES (?)",
        (title,),  # Use ? placeholder — NEVER use f-strings!
    )
    conn.commit()
    return cursor.lastrowid

task_id = insert_task(conn, "Learn Python")
print(f"Created task with ID: {task_id}")

Important: Always use ? placeholders for values. Never put user input directly in SQL strings. This prevents SQL injection attacks:

# DANGEROUS — SQL injection vulnerability
title = "'; DROP TABLE tasks; --"
conn.execute(f"INSERT INTO tasks (title) VALUES ('{title}')")  # BAD!

# SAFE — parameterized query
conn.execute("INSERT INTO tasks (title) VALUES (?)", (title,))  # GOOD!

Query Data

def get_all_tasks(conn: sqlite3.Connection) -> list[dict]:
    """Get all tasks."""
    cursor = conn.execute("SELECT * FROM tasks ORDER BY id")
    return [dict(row) for row in cursor.fetchall()]

def get_task_by_id(conn: sqlite3.Connection, task_id: int) -> dict | None:
    """Get a single task by ID."""
    cursor = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,))
    row = cursor.fetchone()
    return dict(row) if row else None

tasks = get_all_tasks(conn)
for task in tasks:
    status = "done" if task["done"] else "pending"
    print(f"[{task['id']}] {task['title']} ({status})")

Update Data

def mark_task_done(conn: sqlite3.Connection, task_id: int) -> bool:
    """Mark a task as done. Returns True if task existed."""
    cursor = conn.execute(
        "UPDATE tasks SET done = 1 WHERE id = ?",
        (task_id,),
    )
    conn.commit()
    return cursor.rowcount > 0

success = mark_task_done(conn, 1)
print(f"Marked as done: {success}")

Delete Data

def delete_task(conn: sqlite3.Connection, task_id: int) -> bool:
    """Delete a task. Returns True if task existed."""
    cursor = conn.execute(
        "DELETE FROM tasks WHERE id = ?",
        (task_id,),
    )
    conn.commit()
    return cursor.rowcount > 0

Search Data

def search_tasks(conn: sqlite3.Connection, keyword: str) -> list[dict]:
    """Search tasks by title."""
    cursor = conn.execute(
        "SELECT * FROM tasks WHERE title LIKE ?",
        (f"%{keyword}%",),
    )
    return [dict(row) for row in cursor.fetchall()]

results = search_tasks(conn, "Python")
print(f"Found {len(results)} tasks matching 'Python'")

Close the Connection

Always close the connection when done:

conn.close()

Or better, use a context manager (from Tutorial #15):

with sqlite3.connect("tasks.db") as conn:
    conn.row_factory = sqlite3.Row
    tasks = get_all_tasks(conn)
# Transaction is committed automatically on success (rolled back on exception)
# Note: the connection is NOT closed — call conn.close() if needed

Unlike file objects, the sqlite3 context manager only manages the transaction — it commits on success and rolls back if an exception occurs inside the with block. It does not close the connection. If you need the connection to be closed after the block, call conn.close() explicitly after the with statement.

Part 2: SQLAlchemy — The Python ORM

Writing SQL strings by hand works for simple cases, but it gets messy for complex applications. An ORM (Object-Relational Mapper) lets you work with Python classes instead of SQL strings.

SQLAlchemy is the most popular Python ORM. It supports SQLite, PostgreSQL, MySQL, and many other databases.

Install it:

pip install sqlalchemy

Defining Models

A model is a Python class that represents a database table:

from sqlalchemy import String, ForeignKey, create_engine
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
    sessionmaker,
)

class Base(DeclarativeBase):
    """Base class for all models."""
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200), unique=True)

    # A user has many posts
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan",
    )

    def __repr__(self) -> str:
        return f"User(id={self.id}, name='{self.name}')"

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String(5000), default="")
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # A post belongs to a user
    author: Mapped["User"] = relationship(back_populates="posts")

    def __repr__(self) -> str:
        return f"Post(id={self.id}, title='{self.title}')"

This uses SQLAlchemy 2.0 style with Mapped type hints. Each class attribute becomes a column in the database.

Creating the Database

# Create an in-memory SQLite database
engine = create_engine("sqlite:///:memory:", echo=False)

# Create all tables from our models
Base.metadata.create_all(engine)

# Create a session factory
SessionLocal = sessionmaker(bind=engine)
session = SessionLocal()

For a file-based database, use "sqlite:///myapp.db". For PostgreSQL: "postgresql://user:pass@localhost/dbname".

Create (INSERT)

# Create a user
alex = User(name="Alex", email="alex@example.com")
session.add(alex)
session.commit()
session.refresh(alex)  # Load the auto-generated ID

print(alex.id)    # 1
print(alex.name)  # Alex

Read (SELECT)

# Get by ID
user = session.get(User, 1)
print(user)  # User(id=1, name='Alex')

# Get by email
user = session.query(User).filter(User.email == "alex@example.com").first()

# Get all users
all_users = session.query(User).all()

# Search
results = session.query(User).filter(User.name.like("%Alex%")).all()

Update

user = session.get(User, 1)
user.name = "Alex Updated"
session.commit()

Delete

user = session.get(User, 1)
session.delete(user)
session.commit()

Complete CRUD Functions

Here is a clean pattern for organizing CRUD operations:

from sqlalchemy.orm import Session

def create_user(session: Session, name: str, email: str) -> User:
    user = User(name=name, email=email)
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

def get_user_by_id(session: Session, user_id: int) -> User | None:
    return session.get(User, user_id)

def get_all_users(session: Session) -> list[User]:
    return session.query(User).all()

def update_user(session: Session, user_id: int, name: str | None = None) -> User | None:
    user = session.get(User, user_id)
    if user is None:
        return None
    if name is not None:
        user.name = name
    session.commit()
    session.refresh(user)
    return user

def delete_user(session: Session, user_id: int) -> bool:
    user = session.get(User, user_id)
    if user is None:
        return False
    session.delete(user)
    session.commit()
    return True

Relationships

Relationships connect tables together. In our example:

  • A User has many Posts (one-to-many)
  • A Post belongs to one User
# Create a user and posts
alex = create_user(session, "Alex", "alex@example.com")
post1 = Post(title="Python Basics", content="...", author_id=alex.id)
post2 = Post(title="Python ORM", content="...", author_id=alex.id)
session.add_all([post1, post2])
session.commit()

# Access posts through the user
print(f"Alex has {len(alex.posts)} posts")
for post in alex.posts:
    print(f"  - {post.title}")

# Access user through the post
print(f"Post author: {post1.author.name}")

Output:

Alex has 2 posts
  - Python Basics
  - Python ORM
Post author: Alex

The relationship() function tells SQLAlchemy how the tables connect. back_populates creates a two-way link — you can navigate from User to Posts and from Post to User.

Cascade Delete

When you delete a user, what happens to their posts? The cascade="all, delete-orphan" setting means: delete the user’s posts too.

delete_user(session, alex.id)
# All of Alex's posts are also deleted

Without cascade, deleting a user with posts would cause a foreign key error.

Transactions

A transaction groups multiple operations. Either all succeed or all fail:

def transfer_posts(session: Session, from_id: int, to_id: int) -> int:
    """Transfer all posts from one user to another."""
    from_user = session.get(User, from_id)
    to_user = session.get(User, to_id)

    if from_user is None or to_user is None:
        raise ValueError("Both users must exist")

    count = 0
    for post in from_user.posts:
        post.author_id = to_id
        count += 1

    session.commit()  # All changes saved at once
    return count

If commit() is never called (e.g., an exception is raised), uncommitted changes stay in memory but are not written to the database. For explicit rollback on error, use session.rollback() in an except block or wrap the logic in with session.begin(): which commits on success and rolls back on exception.

Database Migrations with Alembic

As your application grows, you will need to change your database schema (add columns, rename tables, etc.). Alembic is the migration tool for SQLAlchemy:

pip install alembic
alembic init migrations
alembic revision --autogenerate -m "Add users table"
alembic upgrade head

Alembic tracks schema changes and applies them in order. This is essential for production applications. We will not cover Alembic in detail here, but know that it exists and you will need it.

Common Mistakes

1. SQL Injection

# DANGEROUS — never do this
name = user_input
conn.execute(f"SELECT * FROM users WHERE name = '{name}'")

# SAFE — use parameterized queries
conn.execute("SELECT * FROM users WHERE name = ?", (name,))

2. Forgetting to Close Connections

# BAD — connection stays open
conn = sqlite3.connect("db.sqlite3")
data = conn.execute("SELECT * FROM users").fetchall()
# Forgot conn.close()!

# GOOD — use try/finally to ensure cleanup
conn = sqlite3.connect("db.sqlite3")
try:
    data = conn.execute("SELECT * FROM users").fetchall()
finally:
    conn.close()

3. N+1 Query Problem

# BAD — one query per user (N+1 queries)
users = session.query(User).all()
for user in users:
    print(len(user.posts))  # Each access triggers a new query!

# BETTER — eager loading
from sqlalchemy.orm import joinedload
users = session.query(User).options(joinedload(User.posts)).all()
for user in users:
    print(len(user.posts))  # No extra queries

Source Code

You can find all the code from this tutorial on GitHub:

GitHub: python-tutorial/tutorial-20-databases

What’s Next?

In the next tutorial, we will learn about logging and debugging — how to replace print() with proper logging, use breakpoint() for debugging, and profile your code.