Database Programming in Python | Python Course | Dataplexa

Database Programming in Python

Almost every real application stores data that outlives the program — user accounts, orders, logs, content. Databases are the standard solution. Python gives you two main paths: the built-in sqlite3 module for lightweight, file-based databases with no setup, and SQLAlchemy — the most widely used Python database toolkit — for working with any database engine through a clean, Pythonic interface.

This lesson covers both: raw SQL with sqlite3, SQLAlchemy Core for flexible query building, and SQLAlchemy ORM for mapping Python classes directly to database tables.

Relational Database Concepts

Table — stores data in rows and columns, like a spreadsheet with strict types.
Row — one record; Column — a named, typed field in every row.
Primary key — a column (usually id) that uniquely identifies each row.
Foreign key — references the primary key of another table, creating a relationship.
Transaction — a group of operations that either all succeed or all fail together (ACID).
SQL — Structured Query Language — the standard for reading and writing relational data.

SQLite with sqlite3 — Create, Insert, Query

SQLite stores an entire database in a single file. No server, no installation, no configuration — import and go. Use ":memory:" for a temporary in-memory database perfect for testing.

import sqlite3

# connect() creates the file if it does not exist
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()

# CREATE TABLE
cursor.execute("""
    CREATE TABLE users (
        id      INTEGER PRIMARY KEY AUTOINCREMENT,
        name    TEXT    NOT NULL,
        email   TEXT    UNIQUE NOT NULL,
        age     INTEGER CHECK(age > 0)
    )
""")

# INSERT — use ? placeholders, NEVER f-strings (SQL injection risk)
users = [
    ("Alice",   "alice@example.com",  30),
    ("Bob",     "bob@example.com",    25),
    ("Charlie", "charlie@example.com", 35),
    ("Diana",   "diana@example.com",  28),
]
cursor.executemany(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)", users
)
conn.commit()   # persist changes

# SELECT with WHERE and ORDER BY
cursor.execute(
    "SELECT id, name, age FROM users WHERE age > ? ORDER BY age DESC",
    (26,)
)
rows = cursor.fetchall()   # list of tuples
for row in rows:
    print(row)

# fetchone() — just the next row
cursor.execute("SELECT COUNT(*) FROM users")
print("Total:", cursor.fetchone()[0])

conn.close()

(3, 'Charlie', 35) (1, 'Alice', 30) (4, 'Diana', 28) Total: 4

Always use ? placeholders — never string-format values into SQL. Doing so opens a SQL injection vulnerability.
conn.commit() saves changes — without it, inserts/updates/deletes are not persisted.
cursor.fetchall() — all rows as tuples; cursor.fetchone() — next single row; cursor.fetchmany(n) — next n rows.
cursor.executemany(sql, list_of_tuples) — inserts multiple rows efficiently in one call.

UPDATE, DELETE, and the Context Manager

The with conn: block automatically commits on success and rolls back on exception — safer than manual commit/rollback.

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("""
    CREATE TABLE products (
        id    INTEGER PRIMARY KEY AUTOINCREMENT,
        name  TEXT    NOT NULL,
        price REAL    NOT NULL
    )
""")

# with conn: auto-commits on success, auto-rollbacks on exception
with conn:
    conn.execute("INSERT INTO products (name, price) VALUES (?, ?)", ("Notebook", 4.99))
    conn.execute("INSERT INTO products (name, price) VALUES (?, ?)", ("Pen",      1.50))
    conn.execute("INSERT INTO products (name, price) VALUES (?, ?)", ("Desk",    89.99))

# UPDATE
with conn:
    conn.execute("UPDATE products SET price = ? WHERE name = ?", (3.99, "Notebook"))

# DELETE
with conn:
    conn.execute("DELETE FROM products WHERE price < ?", (2.00,))

# Row factory — access columns by name instead of index
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM products ORDER BY price")

for row in cursor.fetchall():
    print(f"{row['name']:10} ${row['price']:.2f}")

# rowcount — how many rows were affected by last operation
cursor.execute("UPDATE products SET price = price * 1.1")
print(f"
Updated {cursor.rowcount} rows")

conn.close()

Notebook $3.99 Desk $89.99 Updated 2 rows

conn.row_factory = sqlite3.Row — access columns by name (row["name"]) instead of position (row[0]).
cursor.rowcount — number of rows affected by the last INSERT, UPDATE, or DELETE.
The with conn: context manager handles the transaction — it does not close the connection.

SQLAlchemy Core — SQL with Python Objects

SQLAlchemy is the standard Python database library for production applications. The Core layer is close to raw SQL but uses Python objects — and crucially, swapping the connection string switches the entire database engine without touching any query code.

# pip install sqlalchemy
from sqlalchemy import create_engine, text

# Connection string — swap for any database without changing query code
# PostgreSQL: "postgresql://user:password@localhost/dbname"
# MySQL:      "mysql+pymysql://user:password@localhost/dbname"
engine = create_engine("sqlite:///:memory:", echo=False)

with engine.connect() as conn:
    conn.execute(text("""
        CREATE TABLE employees (
            id         INTEGER PRIMARY KEY,
            name       TEXT    NOT NULL,
            department TEXT,
            salary     REAL
        )
    """))
    conn.execute(text("""
        INSERT INTO employees (name, department, salary) VALUES
        ('Alice',   'Engineering', 95000),
        ('Bob',     'Marketing',   72000),
        ('Charlie', 'Engineering', 88000),
        ('Diana',   'HR',          65000)
    """))
    conn.commit()

# Named parameters with :name syntax
with engine.connect() as conn:
    result = conn.execute(
        text("SELECT name, salary FROM employees WHERE department = :dept ORDER BY salary DESC"),
        {"dept": "Engineering"}
    )
    for row in result:
        print(f"{row.name}: ${row.salary:,.0f}")

    # Aggregate queries
    avg = conn.execute(text("SELECT AVG(salary) FROM employees")).scalar()
    print(f"
Average salary: ${avg:,.0f}")

Alice: $95,000 Charlie: $88,000 Average salary: $80,000

Use :param_name placeholders with a dict in SQLAlchemy — not ? like bare sqlite3.
.scalar() returns the first column of the first row — useful for aggregates like COUNT, AVG, SUM.
engine.connect() as a context manager auto-closes the connection.
Changing the database is just changing the URL string — all query code stays identical.

SQLAlchemy ORM — Classes as Tables

The ORM maps Python classes to database tables. Define your schema once as Python classes and interact with the database entirely through objects — no SQL strings required for most operations.

from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Session, relationship

engine = create_engine("sqlite:///:memory:")

class Base(DeclarativeBase):
    pass

class Category(Base):
    __tablename__ = "categories"
    id   = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False, unique=True)
    products = relationship("Product", back_populates="category_obj")
    def __repr__(self): return f"Category({self.name!r})"

class Product(Base):
    __tablename__ = "products"
    id          = Column(Integer, primary_key=True, autoincrement=True)
    name        = Column(String, nullable=False)
    price       = Column(Float,  nullable=False)
    category_id = Column(Integer, ForeignKey("categories.id"))
    category_obj = relationship("Category", back_populates="products")
    def __repr__(self): return f"Product({self.name!r}, ${self.price:.2f})"

Base.metadata.create_all(engine)

with Session(engine) as session:
    elec = Category(name="Electronics")
    stat = Category(name="Stationery")
    session.add_all([elec, stat])
    session.flush()   # assigns IDs without committing

    session.add_all([
        Product(name="Laptop",   price=999.99, category_id=elec.id),
        Product(name="Monitor",  price=349.99, category_id=elec.id),
        Product(name="Notebook", price=4.99,   category_id=stat.id),
        Product(name="Pen",      price=1.50,   category_id=stat.id),
    ])
    session.commit()

with Session(engine) as session:
    # Filter and order
    electronics = (session.query(Product)
                   .filter(Product.category_id == 1)
                   .order_by(Product.price.desc())
                   .all())
    for p in electronics:
        print(p)

    total = session.query(Product).count()
    print(f"
All products: {total}")

Product('Laptop', $999.99) Product('Monitor', $349.99) All products: 4

Each model inherits from Base and uses __tablename__ to name the table.
session.flush() writes pending changes to the database within the transaction but does not commit — useful when you need the generated IDs before committing.
ForeignKey("categories.id") + relationship() connects models and enables Python-level navigation.
session.rollback() cancels all staged changes since the last commit.

ORM — Update, Delete, and Bulk Operations

with Session(engine) as session:
    # Update — fetch, modify attribute, commit
    laptop = session.query(Product).filter(Product.name == "Laptop").first()
    if laptop:
        laptop.price = 899.99
        session.commit()
        print("Updated:", laptop)

    # Bulk update — no need to load objects
    updated = (session.query(Product)
               .filter(Product.category_id == 2)
               .update({"price": Product.price * 1.1}))
    session.commit()
    print(f"Bulk updated {updated} stationery products")

    # Delete — fetch and delete
    pen = session.query(Product).filter(Product.name == "Pen").first()
    if pen:
        session.delete(pen)
        session.commit()
        print("Deleted:", pen.name)

    # Verify
    remaining = session.query(Product).order_by(Product.price.desc()).all()
    print("Remaining:", remaining)

Updated: Product('Laptop', $899.99) Bulk updated 2 stationery products Deleted: Pen Remaining: [Product('Laptop', $899.99), Product('Monitor', $349.99), Product('Notebook', $5.49)]

Quick Reference Table

Tool	Best For	Key Usage
`sqlite3`	Scripts, prototypes, local storage	`cursor.execute(sql, params)`
SQLAlchemy Core	Flexible SQL, multi-DB apps	`conn.execute(text(sql), params)`
SQLAlchemy ORM	Applications, web apps, complex models	`session.query(Model).filter(...)`
`conn.row_factory`	Named column access in sqlite3	`conn.row_factory = sqlite3.Row`
`?` / `:name`	Safe parameter binding	Always use — never f-string SQL values
`session.flush()`	Write to DB without committing	Use when you need generated IDs early

Practice

Why should you never use f-strings to insert values into SQL queries?

What does conn.commit() do in sqlite3?

What does setting conn.row_factory = sqlite3.Row enable?

In SQLAlchemy ORM, which method stages multiple new objects for insertion in one call?

What is the main advantage of SQLAlchemy over bare sqlite3 for production applications?

Which SQLAlchemy ORM method writes pending changes to the database without committing the transaction?

Quick Quiz

What connection string creates an in-memory SQLite database in SQLAlchemy?

sqlite:///:memory:
:memory:
sqlite:///ram

What happens inside a with conn: block in sqlite3 if an exception is raised?

The transaction is automatically rolled back
The transaction is committed regardless
The exception is suppressed

In SQLAlchemy ORM, what class attribute tells SQLAlchemy which table a model maps to?

__tablename__
__table__
table_name

What is the placeholder syntax for named parameters in SQLAlchemy's text() queries?

:param_name
?
%(param_name)s

Which method creates all tables defined by ORM models in the database?

Base.metadata.create_all(engine)
engine.create_tables()
session.create_all()

Which sqlite3 cursor attribute tells you how many rows were affected by the last INSERT, UPDATE, or DELETE?

cursor.rowcount
cursor.affected
cursor.count

NEXT UP

CSV & XML Files in Python

Reading, writing, and parsing structured data files — Python's csv module, DictReader/DictWriter, and XML with xml.etree.ElementTree.

← Previous Course Index Next →