Introduction to SQLAlchemy Architecture
SQLAlchemy is the most widely used SQL toolkit and Object-Relational Mapping (ORM) library in the Python ecosystem. Since its inception in 2005, it has evolved into a mature, battle-tested framework that powers applications at companies like Reddit, Yelp, and Dropbox. However, its flexibility and power can also be its downfall: without a well-thought-out architecture, SQLAlchemy projects quickly devolve into tangled webs of session leaks, circular imports, and untestable code.
This tutorial explores the architectural patterns and project structures that make SQLAlchemy applications maintainable, scalable, and testable. We'll cover the core building blocks of SQLAlchemy, examine proven design patterns, and walk through a production-ready project layout that you can adapt to your own applications.
Understanding SQLAlchemy's Core Components
Before diving into architecture, it's essential to understand the layers SQLAlchemy provides. SQLAlchemy is not a single monolithic ORM — it's a layered toolkit with two primary APIs: Core and ORM. Understanding how these layers interact is the foundation of any good SQLAlchemy architecture.
The Core Layer
Core is the lower-level layer that provides a SQL expression language, connection pooling, type coercion, and schema definition. It operates on tables and columns rather than Python objects. Core is ideal for scenarios where you need fine-grained control over SQL generation or where the overhead of ORM object tracking is unnecessary.
The ORM Layer
The ORM layer sits on top of Core and maps Python classes to database tables. It provides the declarative mapping system, the identity map, the unit of work pattern, and lazy loading. The ORM is what most developers interact with daily, but it relies entirely on Core under the hood.
Key Architectural Objects
Several objects form the backbone of any SQLAlchemy application. Understanding their roles and lifecycles is critical:
Engine— The starting point for any SQLAlchemy application. It holds the connection pool and dialect, and acts as the factory for database connections.Session— The ORM's primary interface for database operations. It maintains the identity map and tracks changes to objects.Declarative Base— The registry that maps Python classes to database tables. In modern SQLAlchemy (2.0+), this is managed byDeclarativeBaseorregistry.MetaData— A collection ofTableobjects that describes the database schema. It's used for schema creation and reflection.SessionMaker— A factory class that producesSessioninstances with consistent configuration.
Why Architecture Matters
SQLAlchemy's flexibility means there are many ways to structure an application, and most of them are wrong for production use. Poor architecture leads to several common problems:
- Session leaks — Sessions that are never closed exhaust the connection pool and cause application hangs.
- Circular imports — Models that reference each other, or models imported by configuration code that also imports models, create import cycles.
- Untestable code — When database sessions are created inline or passed through global state, unit testing becomes nearly impossible.
- Thread safety issues — Sessions are not thread-safe. Sharing a session across requests in a web application causes data corruption.
- Coupled business logic — When ORM models contain business logic, they become difficult to test and reuse outside the database context.
A well-designed SQLAlchemy architecture addresses all of these concerns by establishing clear boundaries, managing session lifecycles explicitly, and separating concerns between data access, business logic, and presentation.
The Layered Architecture Pattern
The most robust pattern for SQLAlchemy applications is a layered architecture. Each layer has a single responsibility and depends only on the layer below it. This separation makes the codebase testable, maintainable, and adaptable to changing requirements.
Layer Overview
- Models Layer — Defines the database schema and ORM mappings. Contains no business logic.
- Data Access Layer (Repositories) — Encapsulates all database queries. Provides a clean API for retrieving and persisting entities.
- Service Layer — Contains business logic. Orchestrates repository calls and enforces business rules.
- Presentation/API Layer — Handles HTTP requests, input validation, and response formatting. Calls services, never repositories directly.
Project Structure
Here is a production-ready project structure that implements the layered architecture:
myapp/
├── alembic/ # Database migrations
│ ├── versions/
│ └── env.py
├── alembic.ini
├── pyproject.toml
├── tests/
│ ├── conftest.py
│ ├── unit/
│ └── integration/
└── src/
└── myapp/
├── __init__.py
├── config.py # Application configuration
├── database.py # Engine, session factory, base class
├── models/ # ORM models (schema definitions)
│ ├── __init__.py
│ ├── base.py
│ ├── user.py
│ ├── post.py
│ └── association.py
├── repositories/ # Data access layer
│ ├── __init__.py
│ ├── base.py
│ ├── user_repository.py
│ └── post_repository.py
├── services/ # Business logic layer
│ ├── __init__.py
│ ├── user_service.py
│ └── post_service.py
├── schemas/ # Pydantic / serialization schemas
│ ├── __init__.py
│ ├── user_schema.py
│ └── post_schema.py
├── api/ # Presentation layer
│ ├── __init__.py
│ ├── routes/
│ │ ├── __init__.py
│ │ ├── user_routes.py
│ │ └── post_routes.py
│ └── dependencies.py
└── main.py # Application entry point
This structure keeps each layer in its own package, making dependencies explicit and the codebase navigable. Let's now examine each layer in detail.
Setting Up the Database Foundation
The database.py module is the heart of the application's SQLAlchemy setup. It creates the engine, the session factory, and the declarative base. Centralizing these objects prevents circular imports and ensures consistent configuration across the application.
# src/myapp/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session
from typing import Generator
from myapp.config import settings
class Base(DeclarativeBase):
"""Declarative base class for all ORM models."""
pass
# The engine is created once at module load time.
# It manages the connection pool internally.
engine = create_engine(
settings.database_url,
pool_size=10,
max_overflow=20,
pool_pre_ping=True,
pool_recycle=3600,
echo=settings.debug,
)
# SessionMaker is a factory that produces Session instances.
# Each call to SessionLocal() creates a new, independent session.
SessionLocal = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False,
expire_on_commit=False,
)
def get_db() -> Generator[Session, None, None]:
"""
Dependency that yields a database session and ensures it is closed.
Use this in web frameworks (FastAPI, Flask, etc.) to manage session
lifecycle per request.
"""
db = SessionLocal()
try:
yield db
finally:
db.close()
Several design decisions in this module deserve explanation. The pool_pre_ping=True option enables connection health checks before each checkout, preventing errors from stale connections. The expire_on_commit=False setting prevents objects from being expired after a commit, which is important when you need to return objects from an API after persisting them. The get_db function is a generator-based dependency that guarantees session cleanup even if an exception occurs.
Defining Models
Models should be pure schema definitions. They describe the structure of your data and the relationships between tables, but they should not contain business logic. Keeping models thin makes them easier to test and prevents the model layer from becoming a dumping ground for unrelated functionality.
The Base Class
# src/myapp/models/base.py
from datetime import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from myapp.database import Base
class TimestampMixin:
"""Mixin that adds created_at and updated_at columns."""
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)
Model Definitions
# src/myapp/models/user.py
from typing import List, Optional
from sqlalchemy import String, Boolean, Integer
from sqlalchemy.orm import Mapped, mapped_column, relationship
from myapp.database import Base
from myapp.models.base import TimestampMixin
from myapp.models.association import user_post_association
class User(Base, TimestampMixin):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
username: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, index=True)
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
full_name: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
is_superuser: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
# Relationship to posts — note we use string references to avoid
# circular import issues at the Python level.
posts: Mapped[List["Post"]] = relationship(
"Post",
secondary=user_post_association,
back_populates="authors",
lazy="selectin",
)
def __repr__(self) -> str:
return f"<User(id={self.id}, email={self.email!r})>"
# src/myapp/models/post.py
from typing import List, Optional
from sqlalchemy import String, Text, Integer, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from myapp.database import Base
from myapp.models.base import TimestampMixin
from myapp.models.association import user_post_association
class Post(Base, TimestampMixin):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column(String(300), nullable=False)
body: Mapped[str] = mapped_column(Text, nullable=False)
views: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
authors: Mapped[List["User"]] = relationship(
"User",
secondary=user_post_association,
back_populates="posts",
lazy="selectin",
)
def __repr__(self) -> str:
return f"<Post(id={self.id}, title={self.title!r})>"
# src/myapp/models/association.py
from sqlalchemy import Table, Column, Integer, ForeignKey
from myapp.database import Base
# Association table for the many-to-many relationship between users and posts.
# This is defined separately to avoid circular references between model files.
user_post_association = Table(
"user_post_association",
Base.metadata,
Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
Column("post_id", Integer, ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True),
)
# src/myapp/models/__init__.py
from myapp.models.user import User
from myapp.models.post import Post
from myapp.models.association import user_post_association
__all__ = ["User", "Post", "user_post_association"]
The __init__.py file in the models package serves a dual purpose. It provides a clean import surface for the rest of the application, and it ensures that all models are imported (and thus registered with the declarative base metadata) before any schema operations like create_all or Alembic migrations are run.
The Repository Pattern
The repository pattern is the cornerstone of a clean data access layer. A repository encapsulates all database queries for a specific entity, providing a collection-like interface for retrieving and persisting objects. This abstraction has several benefits: it centralizes query logic, makes mocking easy for unit tests, and allows you to swap the data source without touching the service layer.
Base Repository
# src/myapp/repositories/base.py
from typing import Generic, TypeVar, Type, Optional, List, Any
from sqlalchemy import select, func
from sqlalchemy.orm import Session
from myapp.database import Base
ModelType = TypeVar("ModelType", bound=Base)
class BaseRepository(Generic[ModelType]):
"""Generic repository providing common CRUD operations."""
def __init__(self, model: Type[ModelType], db: Session):
self.model = model
self.db = db
def get_by_id(self, id: int) -> Optional[ModelType]:
stmt = select(self.model).where(self.model.id == id)
result = self.db.execute(stmt)
return result.scalar_one_or_none()
def get_all(self, skip: int = 0, limit: int = 100) -> List[ModelType]:
stmt = select(self.model).offset(skip).limit(limit)
result = self.db.execute(stmt)
return list(result.scalars().all())
def count(self) -> int:
stmt = select(func.count()).select_from(self.model)
result = self.db.execute(stmt)
return result.scalar_one()
def create(self, **kwargs: Any) -> ModelType:
instance = self.model(**kwargs)
self.db.add(instance)
self.db.flush() # Flush to get the ID without committing
self.db.refresh(instance)
return instance
def update(self, instance: ModelType, **kwargs: Any) -> ModelType:
for key, value in kwargs.items():
setattr(instance, key, value)
self.db.flush()
self.db.refresh(instance)
return instance
def delete(self, instance: ModelType) -> None:
self.db.delete(instance)
self.db.flush()
def exists(self, id: int) -> bool:
stmt = select(self.model.id).where(self.model.id == id).limit(1)
result = self.db.execute(stmt)
return result.scalar_one_or_none() is not None
Concrete Repositories
# src/myapp/repositories/user_repository.py
from typing import Optional, List
from sqlalchemy import select, or_
from sqlalchemy.orm import Session
from myapp.models.user import User
from myapp.repositories.base import BaseRepository
class UserRepository(BaseRepository[User]):
def __init__(self, db: Session):
super().__init__(User, db)
def get_by_email(self, email: str) -> Optional[User]:
stmt = select(User).where(User.email == email)
result = self.db.execute(stmt)
return result.scalar_one_or_none()
def get_by_username(self, username: str) -> Optional[User]:
stmt = select(User).where(User.username == username)
result = self.db.execute(stmt)
return result.scalar_one_or_none()
def search(self, query: str, skip: int = 0, limit: int = 20) -> List[User]:
pattern = f"%{query}%"
stmt = (
select(User)
.where(
or_(
User.username.ilike(pattern),
User.email.ilike(pattern),
User.full_name.ilike(pattern),
)
)
.offset(skip)
.limit(limit)
)
result = self.db.execute(stmt)
return list(result.scalars().all())
def get_active_users(self, skip: int = 0, limit: int = 100) -> List[User]:
stmt = (
select(User)
.where(User.is_active == True)
.offset(skip)
.limit(limit)
)
result = self.db.execute(stmt)
return list(result.scalars().all())
# src/myapp/repositories/post_repository.py
from typing import Optional, List
from sqlalchemy import select, desc
from sqlalchemy.orm import Session
from myapp.models.post import Post
from myapp.models.user import User
from myapp.repositories.base import BaseRepository
class PostRepository(BaseRepository[Post]):
def __init__(self, db: Session):
super().__init__(Post, db)
def get_by_author(self, user_id: int, skip: int = 0, limit: int = 20) -> List[Post]:
stmt = (
select(Post)
.join(Post.authors)
.where(User.id == user_id)
.offset(skip)
.limit(limit)
)
result = self.db.execute(stmt)
return list(result.scalars().all())
def get_recent(self, limit: int = 10) -> List[Post]:
stmt = select(Post).order_by(desc(Post.created_at)).limit(limit)
result = self.db.execute(stmt)
return list(result.scalars().all())
def get_popular(self, min_views: int = 100, limit: int = 10) -> List[Post]:
stmt = (
select(Post)
.where(Post.views >= min_views)
.order_by(desc(Post.views))
.limit(limit)
)
result = self.db.execute(stmt)
return list(result.scalars().all())
def increment_views(self, post_id: int) -> Optional[Post]:
post = self.get_by_id(post_id)
if post:
post.views += 1
self.db.flush()
self.db.refresh(post)
return post
Notice that repositories use flush() rather than commit(). This is a deliberate design choice: the repository layer should not control transaction boundaries. Transaction management is the responsibility of the service layer, which coordinates multiple repository operations within a single unit of work.
The Service Layer
The service layer contains business logic and transaction management. Services orchestrate repository calls, enforce business rules, and commit or roll back transactions. This is where domain logic lives, keeping it separate from both the database access code and the HTTP layer.
# src/myapp/services/user_service.py
from typing import Optional, List
from sqlalchemy.orm import Session
from myapp.repositories.user_repository import UserRepository
from myapp.repositories.post_repository import PostRepository
from myapp.models.user import User
class UserService:
def __init__(self, db: Session):
self.db = db
self.user_repo = UserRepository(db)
self.post_repo = PostRepository(db)
def register_user(
self,
email: str,
username: str,
password: str,
full_name: Optional[str] = None,
) -> User:
# Business rule: email and username must be unique
if self.user_repo.get_by_email(email):
raise ValueError(f"Email {email} is already registered")
if self.user_repo.get_by_username(username):
raise ValueError(f"Username {username} is already taken")
# In a real application, hash the password here
hashed_password = self._hash_password(password)
user = self.user_repo.create(
email=email,
username=username,
hashed_password=hashed_password,
full_name=full_name,
is_active=True,
is_superuser=False,
)
self.db.commit()
self.db.refresh(user)
return user
def deactivate_user(self, user_id: int) -> Optional[User]:
user = self.user_repo.get_by_id(user_id)
if not user:
return None
if not user.is_active:
raise ValueError("User is already deactivated")
user = self.user_repo.update(user, is_active=False)
self.db.commit()
return user
def get_user_profile(self, user_id: int) -> Optional[dict]:
user = self.user_repo.get_by_id(user_id)
if not user:
return None
post_count = len(user.posts)
return {
"id": user.id,
"email": user.email,
"username": user.username,
"full_name": user.full_name,
"is_active": user.is_active,
"post_count": post_count,
"created_at": user.created_at,
}
def search_users(self, query: str, skip: int = 0, limit: int = 20) -> List[User]:
return self.user_repo.search(query, skip, limit)
def _hash_password(self, password: str) -> str:
# Placeholder — use bcrypt or argon2 in production
import hashlib
return hashlib.sha256(password.encode()).hexdigest()
# src/myapp/services/post_service.py
from typing import Optional, List
from sqlalchemy.orm import Session
from myapp.repositories.post_repository import PostRepository
from myapp.repositories.user_repository import UserRepository
from myapp.models.post import Post
class PostService:
def __init__(self, db: Session):
self.db = db
self.post_repo = PostRepository(db)
self.user_repo = UserRepository(db)
def create_post(
self,
title: str,
body: str,
author_ids: List[int],
) -> Post:
# Business rule: at least one author is required
if not author_ids:
raise ValueError("At least one author is required")
# Validate all authors exist and are active
authors = []
for author_id in author_ids:
user = self.user_repo.get_by_id(author_id)
if not user:
raise ValueError(f"User {author_id} does not exist")
if not user.is_active:
raise ValueError(f"User {author_id} is not active")
authors.append(user)
post = self.post_repo.create(title=title, body=body)
post.authors = authors
self.db.flush()
self.db.commit()
self.db.refresh(post)
return post
def get_post(self, post_id: int) -> Optional[Post]:
return self.post_repo.get_by_id(post_id)
def view_post(self, post_id: int) -> Optional[Post]:
"""Retrieve a post and increment its view count atomically."""
post = self.post_repo.increment_views(post_id)
if post:
self.db.commit()
return post
def get_recent_posts(self, limit: int = 10) -> List[Post]:
return self.post_repo.get_recent(limit)
def get_posts_by_author(self, user_id: int, skip: int = 0, limit: int = 20) -> List[Post]:
return self.post_repo.get_by_author(user_id, skip, limit)
def delete_post(self, post_id: int, requesting_user_id: int) -> bool:
post = self.post_repo.get_by_id(post_id)
if not post:
return False
# Business rule: only authors can delete their posts
author_ids = [a.id for a in post.authors]
if requesting_user_id not in author_ids:
raise PermissionError("Only authors can delete their posts")
self.post_repo.delete(post)
self.db.commit()
return True
Notice how the service layer enforces business rules that span multiple entities. The create_post method validates that all authors exist and are active before creating the post. The delete_post method checks that the requesting user is an author of the post. These rules would be awkward to enforce in the repository layer and impossible to enforce in the model layer.
Wiring It Together: The API Layer
The API layer handles HTTP concerns: parsing requests, calling services, and formatting responses. It should never touch the database directly or contain business logic. Using a dependency injection framework (like FastAPI's Depends) keeps the session lifecycle clean and testable.
# src/myapp/api/dependencies.py
from sqlalchemy.orm import Session
from myapp.database import SessionLocal
from myapp.services.user_service import UserService
from myapp.services.post_service import PostService
def get_db_session():
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_user_service(db: Session = Depends(get_db_session)) -> UserService:
return UserService(db)
def get_post_service(db: Session = Depends(get_db_session)) -> PostService:
return PostService(db)
# src/myapp/api/routes/user_routes.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from myapp.api.dependencies import get_user_service
from myapp.services.user_service import UserService
from myapp.schemas.user_schema import UserCreate, UserResponse, UserSearchResult
router = APIRouter(prefix="/users", tags=["users"])
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(
payload: UserCreate,
user_service: UserService = Depends(get_user_service),
):
try:
user = user_service.register_user(
email=payload.email,
username=payload.username,
password=payload.password,
full_name=payload.full_name,
)
return user
except ValueError as e:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
@router.get("/{user_id}", response_model=UserResponse)
def get_user(
user_id: int,
user_service: UserService = Depends(get_user_service),
):
user = user_service.user_repo.get_by_id(user_id)
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
return user
@router.get("/", response_model=List[UserSearchResult])
def search_users(
q: str,
skip: int = 0,
limit: int = 20,
user_service: UserService = Depends(get_user_service),
):
users = user_service.search_users(q, skip, limit)
return users
@router.patch("/{user_id}/deactivate", response_model=UserResponse)
def deactivate_user(
user_id: int,
user_service: UserService = Depends(get_user_service),
):
try:
user = user_service.deactivate_user(user_id)
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
return user
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
# src/myapp/api/routes/post_routes.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status, Request
from myapp.api.dependencies import get_post_service
from myapp.services.post_service import PostService
from myapp.schemas.post_schema import PostCreate, PostResponse
router = APIRouter(prefix="/posts", tags=["posts"])
@router.post("/", response_model=PostResponse, status_code=status.HTTP_201_CREATED)
def create_post(
payload: PostCreate,
post_service: PostService = Depends(get_post_service),
):
try:
post = post_service.create_post(
title=payload.title,
body=payload.body,
author_ids=payload.author_ids,
)
return post
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
@router.get("/recent", response_model=List[PostResponse])
def get_recent_posts(
limit: int = 10,
post_service: PostService = Depends(get_post_service),
):
return post_service.get_recent_posts(limit)
@router.get("/{post_id}", response_model=PostResponse)
def get_post(
post_id: int,
post_service: PostService = Depends(get_post_service),
):
post = post_service.get_post(post_id)
if not post:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
return post
@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(
post_id: int,
request: Request,
post_service: PostService = Depends(get_post_service),
):
# In a real app, extract user ID from auth token
requesting_user_id = request.headers.get("X-User-ID", type=int)
if not requesting_user_id:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required")
try:
deleted = post_service.delete_post(post_id, requesting_user_id)
if not deleted:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
except PermissionError as e:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
# src/myapp/main.py
from fastapi import FastAPI
from myapp.api.routes import user_routes, post_routes
from myapp.database import Base, engine
app = FastAPI(title="My App", version="1.0.0")
# Register routers
app.include_router(user_routes.router)
app.include_router(post_routes.router)
@app.on_event("startup")
def startup():
# In production, use Alembic migrations instead of create_all
Base.metadata.create_all(bind=engine)
@app.get("/health")
def health_check():
return {"status": "healthy"}
Serialization Schemas
Serialization schemas (typically Pydantic models in modern Python applications) define the shape of your API data. They serve as a validation layer for incoming data and a serialization layer for outgoing data. Keeping schemas separate from ORM models is critical: ORM models represent database structure, while schemas represent API contracts. These two concerns evolve independently.
# src/myapp/schemas/user_schema.py
from typing import Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field
class UserBase(BaseModel):
email: EmailStr
username: str = Field(..., min_length=3, max_length=100)
full_name: Optional[str] = Field(None, max_length=200)
class UserCreate(UserBase):
password: str = Field(..., min_length=8, max_length=128)
class UserResponse(UserBase):
id: int
is_active: bool
is_superuser: bool
created_at: datetime
model_config = {"from_attributes": True}
class UserSearchResult(BaseModel):
id: int
username: str
email: str
full_name: Optional[str]
model_config = {"from_attributes": True}
# src/myapp/schemas/post_schema.py
from typing import List
from datetime import datetime
from pydantic import BaseModel, Field
class PostCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=300)
body: str = Field(..., min_length=1)
author_ids: List[int] = Field(..., min_length=1)
class PostResponse(BaseModel):
id: int
title: str
body: str
views: int
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
Testing Strategy
A layered architecture makes testing straightforward. Unit tests can mock repositories to test service logic in isolation. Integration tests can use a real database (typically SQLite in-memory or a test PostgreSQL instance) to verify that queries work correctly. The separation of concerns means you can test each layer independently.
# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from myapp.database import Base
from myapp.models import User, Post # Ensure all models are imported
@pytest.fixture(scope="function")
def db_engine():
"""Create a fresh in-memory SQLite database for each test."""
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
yield engine
Base.metadata.drop_all(engine)
engine.dispose()
@pytest.fixture(scope="function")
def db_session(db_engine):
"""Provide a clean database session for each test."""
TestingSessionLocal = sessionmaker(bind=db_engine, autocommit=False, autoflush=False)
session = TestingSessionLocal()
try:
yield session
finally:
session.close()
# tests/unit/test_user_service.py
from unittest.mock import MagicMock, patch
import pytest
from myapp.services.user_service import UserService
def test_register_user_raises_on_duplicate_email():
db = MagicMock()
service = UserService(db)
# Mock the repository to simulate an existing user
service.user_repo = MagicMock()
service.user_repo.get_by_email.return_value = MagicMock()
with pytest.raises(ValueError, match="already registered"):
service.register_user(
email="test@example.com",
username="testuser",
password="securepassword",
)
def test_register_user_success():
db = MagicMock()
service = UserService(db)
service.user_repo = MagicMock()
service.user_repo.get_by_email.return_value = None
service.user_repo.get_by_username.return_value = None
service.user_repo.create.return_value = MagicMock(id=1, email="test@example.com")
user = service.register_user(
email="test@example.com",
username="testuser",
password="securepassword",
)
assert user is not None
service.user_repo.create.assert_called_once()
db.commit.assert_called_once()
# tests/integration/test_post_repository.py
import pytest
from myapp.repositories.user_repository import UserRepository
from myapp.repositories.post_repository import PostRepository
from myapp.models import User, Post
def test_create_and_retrieve_post(db_session):
user_repo = UserRepository(db_session)
post_repo = PostRepository(db_session)
# Create a user
user = user_repo.create(
email="author@example.com",
username="author",
hashed_password="hashed",
is_active=True,
is_superuser=False,
)
db_session.commit()
# Create a post with the user as author
post = post_repo.create(title="Test Post", body="Hello World")
post.authors = [user]
db_session.commit()
db_session.refresh(post)
# Retrieve posts by author
posts = post_repo.get_by_author(user.id)
assert len(posts) == 1
assert posts[0].title == "Test Post"
def test_increment_views(db_session):
post_repo = PostRepository(db_session)
post = post_repo.create(title="Popular Post", body="Viral content")
db_session.commit()
initial_views = post.views
post_repo.increment_views(post.id)
db_session.commit()
updated = post_repo.get_by_id(post.id)
assert updated.views == initial_views + 1
Best Practices
Session Management
Always use a session per request (or per unit of work). Never share sessions across threads or requests. Use the get_db dependency pattern shown above to ensure sessions are properly closed. Avoid global session variables at all costs.
Avoid the N+1 Query Problem
The N+1 query problem is the most common performance issue in ORM applications. It occurs when you load a list of entities and then access a relationship on each entity, triggering a separate query for each one. Use eager loading strategies (selectinload, joinedload, or subqueryload) to load relationships in a single query.
# Bad: N+1 queries
users = db.execute(select(User)).scalars().all()
for user in users:
print(len(user.posts)) # Triggers a query for EACH user
# Good: Single query with eager loading
from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.posts))
users = db.execute(stmt).scalars().all()
for user in users:
print(len(user.posts)) # No