Database Access in Python
Python offers excellent database support through its built-in sqlite3 module for SQLite, the powerful SQLAlchemy ORM for SQL databases, and dedicated drivers for PostgreSQL (psycopg), MySQL (mysqlclient), and NoSQL databases like MongoDB (pymongo) and Redis (redis-py).
SQLite with sqlite3
import sqlite3
from contextlib import closing
# Connect (creates file if not exists)
conn = sqlite3.connect("app.db")
# conn = sqlite3.connect(":memory:") # In-memory database
# Row factory for dict-like access
conn.row_factory = sqlite3.Row
# Create tables
conn.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!)
conn.execute(
"INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
("Alice", "alice@example.com", 30)
)
# Insert many
users = [
("Bob", "bob@example.com", 25),
("Charlie", "charlie@example.com", 35),
("Diana", "diana@example.com", 28),
]
conn.executemany(
"INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
users
)
conn.commit()
# Query data
cursor = conn.execute("SELECT * FROM users WHERE age > ?", (27,))
for row in cursor:
print(f"{row['name']} ({row['email']}): age {row['age']}")
# Fetch methods
cursor = conn.execute("SELECT * FROM users")
first = cursor.fetchone() # Single row
rest = cursor.fetchall() # All remaining rows
# Context manager for automatic commit/rollback
with conn:
conn.execute("UPDATE users SET age = ? WHERE name = ?", (31, "Alice"))
conn.close()
# Best practice: use closing() context manager
with closing(sqlite3.connect("app.db")) as conn:
conn.row_factory = sqlite3.Row
with conn:
rows = conn.execute("SELECT * FROM users").fetchall()
SQLAlchemy ORM
# pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Session, relationship, sessionmaker
# Define the base class
class Base(DeclarativeBase):
pass
# Define models
class Author(Base):
__tablename__ = "authors"
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
email = Column(String(200), unique=True)
books = relationship("Book", back_populates="author", cascade="all, delete-orphan")
def __repr__(self):
return f"Author(id={self.id}, name='{self.name}')"
class Book(Base):
__tablename__ = "books"
id = Column(Integer, primary_key=True)
title = Column(String(200), nullable=False)
price = Column(Float, nullable=False)
author_id = Column(Integer, ForeignKey("authors.id"), nullable=False)
author = relationship("Author", back_populates="books")
def __repr__(self):
return f"Book(id={self.id}, title='{self.title}')"
# Create engine and tables
engine = create_engine("sqlite:///bookstore.db", echo=False)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)
# CRUD operations
with SessionLocal() as session:
# Create
author = Author(name="J.K. Rowling", email="jk@example.com")
author.books.append(Book(title="Harry Potter 1", price=12.99))
author.books.append(Book(title="Harry Potter 2", price=14.99))
session.add(author)
session.commit()
# Read
authors = session.execute(select(Author)).scalars().all()
for a in authors:
print(f"{a.name}: {[b.title for b in a.books]}")
# Query with filters
expensive = session.execute(
select(Book).where(Book.price > 13.00)
).scalars().all()
# Join query
results = session.execute(
select(Book, Author)
.join(Author)
.where(Author.name == "J.K. Rowling")
).all()
# Update
book = session.execute(
select(Book).where(Book.title == "Harry Potter 1")
).scalar_one()
book.price = 15.99
session.commit()
# Delete
session.delete(book)
session.commit()
Async Database Access
# pip install sqlalchemy[asyncio] aiosqlite
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select
# Async engine
async_engine = create_async_engine("sqlite+aiosqlite:///async_app.db")
AsyncSessionLocal = async_sessionmaker(async_engine, class_=AsyncSession)
async def get_users():
async with AsyncSessionLocal() as session:
result = await session.execute(select(User))
users = result.scalars().all()
return users
# With FastAPI
from fastapi import Depends
async def get_db():
async with AsyncSessionLocal() as session:
yield session
@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User))
return result.scalars().all()
Database Migrations with Alembic
# pip install alembic
# Initialize Alembic
# alembic init migrations
# Generate a migration from model changes
# alembic revision --autogenerate -m "add users table"
# Apply migrations
# alembic upgrade head
# Rollback
# alembic downgrade -1
# Example migration file
# def upgrade():
# op.create_table(
# 'users',
# sa.Column('id', sa.Integer(), primary_key=True),
# sa.Column('name', sa.String(100), nullable=False),
# sa.Column('email', sa.String(200), unique=True),
# )
#
# def downgrade():
# op.drop_table('users')
Key Takeaways
- Parameterized queries: Never use f-strings or string formatting in SQL queries
- SQLAlchemy for production: Use the ORM for complex applications
- Async for performance: Use async sessions in async web frameworks
- Alembic for migrations: Track schema changes in version control