Complete CRUD Implementation — From the Data Store to the API Endpoint
CRUD is like the four basic warehouse operations—receiving (Create), inventory counting (Read), stock transfer (Update), and shipping (Delete). The warehouse model ensures that operational processes are standardized, preventing confusion caused by different warehouse managers.
1. What You'll Learn
- Repository Pattern:
ProductRepository/PriceRepositoryEncapsulation - Pagination and Sorting: The Pydantic query models for
limit/offsetandorder_by - Preloading Related Data:
selectinloadvsjoinedload—Performance Trade-offs - Transaction Management: Ensuring Consistency in Cross-Table Operations
- Alice Scenario: Bulk Price Import API—Transactional Processing for Inserting 1,000 Price Records in a Single Request
2. Alice's True Story
(1) Pain Point: SQL scattered throughout routing functions
In Alice's PriceTracker code, SQL queries are written directly within the route handler functions, and there are 30 instances of duplicate select(Product).where(...) logic across 20 endpoints. When she needed to add a "soft delete filter" condition to all product queries, Alice had to make changes in 15 places; missing just two of them caused deleted products to appear in the results.
(2) Solutions for the Warehouse Model
Encapsulate database operations within the Repository class; route functions should only call Repository methods. To add a global filter, you only need to make a single change in the Repository.
PYTHON
class ProductRepository:
async def get_all(self, db: AsyncSession, filters: ProductFilter) -> list[Product]:
stmt = select(Product).where(Product.deleted == False)
# Add filters from Pydantic model
if filters.category:
stmt = stmt.where(Product.category == filters.category)
result = await db.execute(stmt)
return result.scalars().all()
(3) Revenue
SQL logic is centralized in the Repository, making the routing function more concise (5 lines instead of 15). Global filter conditions can be modified in one place to take effect system-wide, so Bob on the front end will no longer see deleted products.
3. Layered Architecture and Data Warehouse Models
(1) Four-layer architecture
flowchart TD
Router[Router Layer] -->|Call| Service[Service Layer]
Service -->|Call| Repository[Repository Layer]
Repository -->|Query| SQLAlchemy[SQLAlchemy ORM]
SQLAlchemy -->|SQL| Database[(PostgreSQL)]
style Router fill:#e3f2fd
style Service fill:#fff3e0
style Repository fill:#e8f5e9
style SQLAlchemy fill:#fce4ec
| Level | Responsibilities | Examples |
|---|---|---|
| Router | Parameter Validation, Response Format | @app.get("/products") |
| Service | Business Logic, Transaction Orchestration | ProductService.create_with_price() |
| Repository | Data Access, SQL Wrappers | ProductRepository.get_by_id() |
| ORM | Object-Relational Mapping | select(Product).where(...) |
(1) ▶ Example: ProductRepository
PYTHON
from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
class ProductRepository:
def __init__(self, db: AsyncSession):
self.db = db
async def get_by_id(self, product_id: int) -> Product | None:
stmt = select(Product).where(Product.id == product_id)
result = await self.db.execute(stmt)
return result.scalar_one_or_none()
async def get_all(
self,
category: str | None = None,
skip: int = 0,
limit: int = 20,
) -> list[Product]:
stmt = select(Product).offset(skip).limit(limit)
if category:
stmt = stmt.where(Product.category == category)
result = await self.db.execute(stmt)
return list(result.scalars().all())
async def create(self, product: ProductCreate) -> Product:
db_product = Product(**product.model_dump())
self.db.add(db_product)
await self.db.flush()
await self.db.refresh(db_product)
return db_product
async def update(self, product_id: int, data: dict) -> Product | None:
stmt = (
update(Product)
.where(Product.id == product_id)
.values(**data)
.returning(Product)
)
result = await self.db.execute(stmt)
return result.scalar_one_or_none()
async def delete(self, product_id: int) -> bool:
stmt = delete(Product).where(Product.id == product_id)
result = await self.db.execute(stmt)
return result.rowcount > 0
Output:
TEXT
# Function defined successfully
(2) ▶ Example: PriceRepository
PYTHON
class PriceRepository:
def __init__(self, db: AsyncSession):
self.db = db
async def get_by_product(
self,
product_id: int,
min_price: float | None = None,
max_price: float | None = None,
skip: int = 0,
limit: int = 50,
) -> list[Price]:
stmt = (
select(Price)
.where(Price.product_id == product_id)
.order_by(Price.recorded_at.desc())
.offset(skip)
.limit(limit)
)
if min_price is not None:
stmt = stmt.where(Price.price >= min_price)
if max_price is not None:
stmt = stmt.where(Price.price <= max_price)
result = await self.db.execute(stmt)
return list(result.scalars().all())
async def bulk_create(self, prices: list[PriceCreate]) -> list[Price]:
db_prices = [Price(**p.model_dump()) for p in prices]
self.db.add_all(db_prices)
await self.db.flush()
return db_prices
Output:
TEXT
# Function defined successfully
4. Pagination and Sorting
(1) Pydantic Query Model
Encapsulate pagination parameters into a PyDantic model to avoid having to declare query parameters repeatedly in each endpoint.
(1) ▶ Example: Paginated Query Model
PYTHON
from pydantic import BaseModel, Field
from typing import Optional
class PaginationParams(BaseModel):
skip: int = Field(0, ge=0, description="Number of records to skip")
limit: int = Field(20, ge=1, le=100, description="Max records to return")
class ProductFilter(PaginationParams):
category: Optional[str] = Field(None, max_length=100)
min_price: Optional[float] = Field(None, ge=0, description="Min price in USD")
max_price: Optional[float] = Field(None, ge=0, description="Max price in USD")
sort_by: str = Field("name", pattern="^(name|base_price|created_at)$")
sort_order: str = Field("asc", pattern="^(asc|desc)$")
Output:
TEXT
# Execution Successful
(2) ▶ Example: Endpoints Using the Pagination Model
PYTHON
from fastapi import FastAPI, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()
@app.get("/products", response_model=list[ProductResponse])
async def list_products(
category: Optional[str] = Query(None),
min_price: Optional[float] = Query(None, ge=0),
max_price: Optional[float] = Query(None, ge=0),
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
):
repo = ProductRepository(db)
return await repo.get_all(
category=category,
min_price=min_price,
max_price=max_price,
skip=skip,
limit=limit,
)
Output:
TEXT
# Function defined successfully
(2) Comparison of Pagination Strategies
| Strategy | Implementation | Advantages | Disadvantages |
|---|---|---|---|
| Offset Pagination | OFFSET n LIMIT m |
Simple, supports page jumping | Poor performance with large offsets |
| Cursor Pagination | WHERE id > cursor LIMIT m |
Performs well with large datasets | Does not support page jumping |
| Keyset Pagination | WHERE created_at < last LIMIT m |
Pagination by sort field works well | Requires an ordered field |
5. Preloading Related Data
(1) The N+1 Query Problem
(1) ▶ Example: Demonstration of the N+1 Problem
PYTHON
# BAD: N+1 queries - 1 query for products + N queries for prices
stmt = select(Product)
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
# Each access triggers a separate query!
print(len(p.prices)) # N queries
Output:
TEXT
# Execution Successful
(2) ▶ Example: selectinload Preloading
PYTHON
from sqlalchemy.orm import selectinload, joinedload
# GOOD: 2 queries total - products + prices (using SELECT IN)
stmt = select(Product).options(selectinload(Product.prices))
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
print(len(p.prices)) # No extra queries
Output:
TEXT
# Execution Successful
(2) selectinload vs joinedload
| Dimension | selectinload | joinedload |
|---|---|---|
| SQL Strategies | SELECT ... WHERE id IN (...) | LEFT OUTER JOIN |
| Number of queries | 2 queries | 1 query |
| Data Volume | Accurate (no duplicate rows) | JOIN may produce duplicate rows |
| Use Cases | One-to-Many (Set) | Many-to-One/One-to-One |
| Performance | Better for large datasets | Better for datasets with few relationships |
| Remove duplicates | Not required | Required unique() |
(3) ▶ Example: Product query with prices preloaded
PYTHON
async def get_product_with_prices(
self, product_id: int
) -> Product | None:
stmt = (
select(Product)
.options(selectinload(Product.prices))
.where(Product.id == product_id)
)
result = await self.db.execute(stmt)
return result.scalar_one_or_none()
Output:
TEXT
# Function defined successfully
6. Transaction Management
(1) Cross-table Transactions
(1) ▶ Example: Bulk Price Import Transaction
PYTHON
from sqlalchemy.ext.asyncio import AsyncSession
class PriceService:
def __init__(self, db: AsyncSession):
self.db = db
self.price_repo = PriceRepository(db)
self.product_repo = ProductRepository(db)
async def bulk_import_prices(
self,
product_id: int,
prices: list[PriceCreate],
) -> list[Price]:
# Verify product exists
product = await self.product_repo.get_by_id(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Validate all prices reference the same product
for p in prices:
p.product_id = product_id
# Bulk insert within transaction (db session handles commit/rollback)
db_prices = await self.price_repo.bulk_create(prices)
# Update product base price to latest
latest_price = db_prices[-1]
await self.product_repo.update(
product_id,
{"base_price": latest_price.price},
)
return db_prices
Output:
TEXT
# Function defined successfully
(2) ▶ Example: Batch Import of Endpoints
PYTHON
@app.post("/products/{product_id}/prices/bulk", response_model=list[PriceResponse])
async def bulk_import(
product_id: int = Path(gt=0),
prices: list[PriceCreate] = ...,
db: AsyncSession = Depends(get_db),
user: dict = Depends(get_current_user),
):
# Subscription check: Free users limited to 1000 prices
if user["subscription"] == "free" and len(prices) > 1000:
raise HTTPException(
status_code=403,
detail="Free plan limited to 1000 prices per import. Upgrade to Pro.",
)
service = PriceService(db)
return await service.bulk_import_prices(product_id, prices)
Output:
TEXT
# Function defined successfully
❓ FAQ
Q Does the Repository have to be a class?
A No, it doesn't have to be; you can also use a function. However, a class encapsulates the DB session state, making it easier to share the session across methods, so the class approach is recommended.
Q What is the difference between
flush and commit?A
flush sends SQL to the database but does not commit the transaction; it can retrieve auto-increment IDs and trigger database constraint checks. commit commits the transaction, making the changes persistent. In yield dependencies, we recommend using flush followed by an automatic commit.Q Does the N+1 problem only occur within loops?
A It mainly occurs when iterating over associated attributes. However, it can also be triggered by PyDantic serialization—if
response_model accesses associated fields. Be sure to preload the data.Q Is there a size limit for
bulk_create?A There is no framework-imposed limit, but PostgreSQL limits the number of bound parameters per operation to 32,767. Inserts in the thousands are no problem, but for tens of thousands, we recommend splitting them into batches.
Q How slow is offset pagination when dealing with large datasets?
A
OFFSET 1000000 It first needs to scan the first 1 million rows before skipping them; the query time is proportional to the offset. For datasets in the millions, we recommend using cursor pagination.Q How do I handle partial failures within a transaction?
A The rollback associated with
yield is automatically triggered when an exception occurs. If you need a partial rollback, use a savepoint (await session.begin_nested()).📖 Summary
- The repository pattern encapsulates SQL operations within the Repository class, so routing functions focus solely on HTTP logic.
- Pagination parameters are encapsulated as a Pydantic model to avoid having to declare query parameters repeatedly for each endpoint
selectinloadis suitable for one-to-many preloading (no duplicate rows);joinedloadis suitable for many-to-one (single query)- Batch operations are executed within a transaction;
yieldrelies on automatic commit/rollback management. - PriceTracker's bulk price import supports thousands of records; Free users are limited to 1,000 records.
📝 Exercises
- Basic Problem (Difficulty ⭐): Implement the
get_by_idandget_allmethods inProductRepository, inject the DB Session into the endpoints, and call them. Hint:select(Product).where(Product.id == product_id) - Advanced Exercise (Difficulty: ⭐⭐): Add support for paginated queries (using the
skipandlimitparameters), implementselectinload(Product.prices)preloading, and return product details that include prices. Hint:options(selectinload(...)) - Challenge (Difficulty ⭐⭐⭐): Implement
PriceService.bulk_import_prices, verify product existence, insert prices in bulk, and update the product'sbase_price—all within a single transaction. Free users are limited to 1,000 records. Hint:bulk_create+update+Depends(get_current_user)
---|
