Databases with Python

1. Overview#

Most real applications need to persist data. Python talks to databases through DB-API drivers (like psycopg2 for PostgreSQL) directly, or through an ORM (Object-Relational Mapper) like SQLAlchemy, which lets you work with Python classes instead of writing raw SQL everywhere. This note covers both, plus async database access — relevant when paired with FastAPI.

---

2. Raw SQL with a DB-API Driver#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import sqlite3

conn = sqlite3.connect("app.db")
cursor = conn.cursor()

cursor.execute("""
 CREATE TABLE IF NOT EXISTS users (
 id INTEGER PRIMARY KEY,
 name TEXT NOT NULL,
 email TEXT UNIQUE
 )
""")

cursor.execute(
 "INSERT INTO users (name, email) VALUES (?, ?)",
 ("Kamal", "kamal@example.com")
)
conn.commit()

cursor.execute("SELECT * FROM users WHERE name = ?", ("Kamal",))
print(cursor.fetchone())

conn.close()

Always use parameterized queries (? placeholders, or %s for PostgreSQL drivers) — never format SQL strings with f-strings or .format(). String-built SQL is the classic entry point for SQL injection attacks.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
# INCORRECT: Never do this
name = "Kamal"
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'") # vulnerable to injection

# CORRECT: Always do this
cursor.execute("SELECT * FROM users WHERE name = ?", (name,))

---

3. SQLAlchemy — The ORM Approach#

3.1 Installation#

bash
Ask AICopy
pip install sqlalchemy

3.2 Defining Models#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
 __tablename__ = "users"

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

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

engine = create_engine("sqlite:///app.db")
Base.metadata.create_all(engine) # creates tables if they don't exist

Session = sessionmaker(bind=engine)

3.3 CRUD Operations#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
session = Session()

# Create
new_user = User(name="Asha", email="asha@example.com")
session.add(new_user)
session.commit()

# Read
user = session.query(User).filter_by(name="Asha").first()
print(user)

all_users = session.query(User).all()

# Update
user.email = "asha.new@example.com"
session.commit()

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

session.close()

3.4 Using a Session as a Context Manager#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from sqlalchemy.orm import Session as SQLASession

with SQLASession(engine) as session:
 session.add(User(name="Ravi", email="ravi@example.com"))
 session.commit()
# Session automatically closed here

---

4. Relationships Between Tables#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Author(Base):
 __tablename__ = "authors"
 id = Column(Integer, primary_key=True)
 name = Column(String)
 books = relationship("Book", back_populates="author")

class Book(Base):
 __tablename__ = "books"
 id = Column(Integer, primary_key=True)
 title = Column(String)
 author_id = Column(Integer, ForeignKey("authors.id"))
 author = relationship("Author", back_populates="books")

# Usage
author = Author(name="R.K. Narayan")
author.books.append(Book(title="Swami and Friends"))
session.add(author)
session.commit()

print(author.books[0].title) # Swami and Friends
print(author.books[0].author.name) # R.K. Narayan

---

5. Querying in Depth#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from sqlalchemy import and_, or_

# Filtering
session.query(User).filter(User.name == "Kamal").all()
session.query(User).filter(User.name.like("%amal%")).all()
session.query(User).filter(and_(User.name == "Kamal", User.id > 1)).all()
session.query(User).filter(or_(User.name == "Kamal", User.name == "Asha")).all()

# Ordering & limiting
session.query(User).order_by(User.name.desc()).limit(10).all()

# Counting
session.query(User).count()

---

6. Async Database Access#

For high-concurrency apps (especially with FastAPI), synchronous DB calls block the event loop. SQLAlchemy supports async drivers for this.

bash
Ask AICopy
pip install sqlalchemy[asyncio] asyncpg # asyncpg for PostgreSQL

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_user_by_id(user_id: int):
 async with AsyncSessionLocal() as session:
 result = await session.get(User, user_id)
 return result

6.1 Using It Inside FastAPI#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from fastapi import FastAPI, Depends

app = FastAPI()

async def get_db():
 async with AsyncSessionLocal() as session:
 yield session

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
 user = await db.get(User, user_id)
 return user

---

7. Connection Pooling#

Opening a new database connection per request is slow — pooling reuses a set of open connections across requests.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from sqlalchemy import create_engine

engine = create_engine(
 "postgresql://user:pass@localhost/mydb",
 pool_size=10, # number of connections kept open
 max_overflow=5, # extra connections allowed under heavy load
 pool_timeout=30, # seconds to wait for a free connection before erroring
 pool_recycle=1800, # recycle connections after 30 min to avoid stale ones
)

---

8. Migrations with Alembic#

Schema changes (adding a column, renaming a table) need to be tracked and applied consistently across environments — that's what Alembic (SQLAlchemy's migration tool) is for.

bash
Ask AICopy
pip install alembic
alembic init migrations
alembic revision --autogenerate -m "add email column to users"
alembic upgrade head

---

9. Common Pitfalls#

INCORRECT: Building SQL Queries with String Formatting#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
cursor.execute(f"SELECT * FROM users WHERE id = {user_id}") # SQL injection risk

CORRECT: Fix — Always Parameterize#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))

INCORRECT: Not Closing Sessions/Connections#

Leaked connections eventually exhaust the connection pool, causing the whole app to hang under load.

CORRECT: Fix — Use Context Managers or Dependency Injection#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
with Session(engine) as session:
 ...
# or, in FastAPI, a dependency with yield handles cleanup automatically

INCORRECT: N+1 Query Problem#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
authors = session.query(Author).all()
for author in authors:
 print(author.books) # triggers a SEPARATE query for every author!

CORRECT: Fix — Eager Loading#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from sqlalchemy.orm import joinedload

authors = session.query(Author).options(joinedload(Author.books)).all()
# Now books are fetched in a single JOIN query instead of one query per author

---

10. Summary & Best Practices Checklist#

• Always use parameterized queries — never build SQL with string interpolation.
• Use SQLAlchemy (or another ORM) for anything beyond trivial scripts — it prevents whole classes of bugs.
• Close sessions/connections properly — use with blocks or dependency injection.
• Watch for the N+1 query problem; use joinedload/selectinload for related data.
• Use async SQLAlchemy + an async driver (asyncpg) when paired with an async framework like FastAPI.
• Configure connection pooling appropriately for your expected load.
• Use Alembic (or an equivalent) to version-control schema changes — never edit production schemas by hand.