Database Access

Database Access

Most real-world applications need to persist data beyond the lifetime of the program. Databases provide structured, reliable, and efficient storage for application data. Python includes built-in support for SQLite through the sqlite3 module and supports virtually every database system through third-party libraries. Understanding how to connect to databases, execute queries, and use an ORM (Object-Relational Mapper) like SQLAlchemy is essential for backend development, data analysis, and building any application that needs persistent storage.

SQLite with the sqlite3 Module

import sqlite3

# Connect (creates file if it doesn't exist)
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()

# Create table
cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# Insert data — always use parameterized queries to prevent SQL injection!
cursor.execute(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    ("Alice", "alice@example.com", 30)
)

# Insert multiple rows
users = [
    ("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()

# Query data
cursor.execute("SELECT * FROM users WHERE age > ?", (26,))
rows = cursor.fetchall()
for row in rows:
    print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}")

# Use Row factory for dict-like access
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
for row in cursor.fetchall():
    print(f"{row['name']}: {row['email']}")

conn.close()

Context Manager Pattern

import sqlite3

def get_connection(db_path="myapp.db"):
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    return conn

# Safe database operations with context manager
def create_user(name, email, age):
    with get_connection() as conn:
        try:
            conn.execute(
                "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
                (name, email, age)
            )
            conn.commit()
            print(f"User {name} created successfully")
        except sqlite3.IntegrityError:
            print(f"User with email {email} already exists")

def get_users(min_age=0):
    with get_connection() as conn:
        cursor = conn.execute(
            "SELECT * FROM users WHERE age >= ? ORDER BY name",
            (min_age,)
        )
        return [dict(row) for row in cursor.fetchall()]

def update_user(user_id, **kwargs):
    fields = ", ".join(f"{k} = ?" for k in kwargs)
    values = list(kwargs.values()) + [user_id]
    with get_connection() as conn:
        conn.execute(f"UPDATE users SET {fields} WHERE id = ?", values)
        conn.commit()

def delete_user(user_id):
    with get_connection() as conn:
        conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
        conn.commit()

SQLAlchemy ORM

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# Setup
engine = create_engine("sqlite:///app.db", echo=False)
Base = declarative_base()
Session = sessionmaker(bind=engine)

# Define model
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(200), unique=True, nullable=False)
    age = Column(Integer)

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

# Create tables
Base.metadata.create_all(engine)

# CRUD operations
session = Session()

# Create
new_user = User(name="Alice", email="alice@example.com", age=30)
session.add(new_user)
session.commit()

# Read
all_users = session.query(User).all()
user = session.query(User).filter_by(name="Alice").first()
young_users = session.query(User).filter(User.age < 30).all()

# Update
user.age = 31
session.commit()

# Delete
session.delete(user)
session.commit()

session.close()

Pro tip: Never build SQL queries by concatenating strings with user input — this creates SQL injection vulnerabilities. Always use parameterized queries (placeholders like ? in sqlite3 or :param in SQLAlchemy) or an ORM to safely handle user data in database queries.

Key Takeaways

• Python's built-in sqlite3 module provides a lightweight database that requires no server setup.
• Always use parameterized queries (? placeholders) to prevent SQL injection attacks.
• Use context managers (with statement) to ensure database connections are properly closed.
• SQLAlchemy ORM maps Python classes to database tables, allowing you to work with objects instead of raw SQL.
• Use conn.row_factory = sqlite3.Row for dictionary-like access to query results.