Phase 2 Comprehensive Exercise — PriceTracker Core Functionality Closed-Loop Implementation
Phase 2 is like the final assembly of a car—the engine (database), steering wheel (routing), seat belts (authentication), and dashboard (middleware) have all been built, and now they're being assembled into a complete car ready to hit the road.
1. What You'll Learn
- Refine the three-tier DI chain:
get_db→get_current_user→require_subscription("pro") - End-to-End JWT Protection: Routing Group Policies for Public Endpoints vs. Authenticated Endpoints
- CORS middleware configuration: Allow Bob's front-end domain to access resources across domains
- End-to-end testing workflow for complete CRUD operations, pagination, and associated data preloading
- Alice Scenario Verification: Pro users can import millions of price data points in bulk, while Free users are limited to 1,000 entries.
2. Alice's True Story
(1) Pain Point: Functions are scattered and cannot be integrated for testing
Alice implemented middleware, dependency injection, the database, CRUD, and JWT authentication separately, but each feature was tested independently. When she tried to integrate all these features, she discovered that the execution order of the CORS middleware and JWT authentication conflicted, the DB session was created twice in the DI chain, and the rate-limiting logic and permission checks for Pro users overrode each other.
(2) A Systematic and Integrated Approach
Phase 2 Comprehensive Practice integrates all modules in the correct hierarchy and order: CORS (outermost layer) → Rate Limiting → Authentication → Business Logic → Database (innermost layer), with each layer having clearly defined responsibilities and operating independently of the others.
(3) Revenue
The integrated PriceTracker core service is fully operational: Bob's frontend successfully accessed cross-domain resources, all protected endpoints require a JWT token, Pro users were imported in bulk without issues, and Free users are correctly rate-limited.
3. Phase 2 Architecture Overview
(1) Complete Request Flow
flowchart TD
Client[Bob Frontend] --> CORS[CORS Middleware]
CORS --> RateLimit[Rate Limit Middleware]
RateLimit --> Router[FastAPI Router]
Router --> AuthDI{Auth Required?}
AuthDI -->|Yes| GetDB[get_db Dependency]
AuthDI -->|No| Handler[Handler]
GetDB --> GetUser[get_current_user]
GetUser --> CheckSub[require_subscription]
CheckSub --> Handler
Handler --> Repo[Repository Layer]
Repo --> DB[(PostgreSQL)]
RateLimit -.->|429| Reject[Rate Limited]
GetUser -.->|401| Unauth[Unauthorized]
CheckSub -.->|403| Forbid[Forbidden]
(2) Routing Group Policy
| Routing Group | Prefix | Authentication | Description |
|---|---|---|---|
| public | / |
None | health, docs |
| api_v1 | /api/v1 |
JWT required | All business endpoints |
| admin | /api/v1/admin |
JWT + Admin | Management Endpoints |
4. Integrating the Complete Code
(1) Application Entry Points and Middleware Configuration
(1) ▶ Example: Complete configuration for main.py
PYTHON
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from app.api.routes import products, prices, auth
from app.core.config import settings
app = FastAPI(
title=settings.app_name,
version="1.0.0",
description="PriceTracker SaaS API - E-commerce price tracking service",
)
# Middleware order: last registered = innermost (executes first on request)
# CORS should be outermost (registered last)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers - public first, then protected
app.include_router(auth.router, tags=["auth"])
app.include_router(
products.router,
prefix="/api/v1",
dependencies=[Depends(get_current_user)],
tags=["products"],
)
app.include_router(
prices.router,
prefix="/api/v1",
dependencies=[Depends(get_current_user)],
tags=["prices"],
)
@app.get("/health")
async def health_check():
return {"status": "healthy", "service": "pricetracker"}
Output:
TEXT
# Function defined successfully
(2) Complete Implementation of a Three-Layer DI Chain
(2) ▶ Example: app/core/deps.py
PYTHON
from fastapi import Depends, HTTPException, Header
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from jose import jwt, JWTError
from app.db import async_session
from app.core.config import settings
from app.core.security import verify_password
from app.models import User
from sqlalchemy import select
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
# Level 1: Database session
async def get_db():
async with async_session() as session:
try:
yield session
except Exception:
await session.rollback()
raise
# Level 2: Current user
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db),
):
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token, settings.secret_key, algorithms=[settings.algorithm]
)
username = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
stmt = select(User).where(User.email == username)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if user is None:
raise credentials_exception
return user
# Level 3: Subscription check
def require_subscription(min_level: str = "free"):
LEVELS = {"free": 0, "pro": 1, "enterprise": 2}
async def check(user: User = Depends(get_current_user)):
if LEVELS.get(user.subscription, 0) < LEVELS.get(min_level, 0):
raise HTTPException(
status_code=403,
detail=f"Requires {min_level} plan. Current: {user.subscription}",
)
return user
return check
Output:
TEXT
# Function defined successfully
(3) Permission Verification Process
flowchart LR
Request[Request with Token] --> ParseToken[Parse JWT Token]
ParseToken --> ValidToken{Token Valid?}
ValidToken -->|No| Return401[Return 401]
ValidToken -->|Yes| QueryUser[Query User from DB]
QueryUser --> UserExists{User Found?}
UserExists -->|No| Return401
UserExists -->|Yes| CheckSub{Subscription Level?}
CheckSub -->|Free| AllowBasic[Allow Basic Endpoints]
CheckSub -->|Pro| AllowPro[Allow Pro Endpoints]
CheckSub -->|Enterprise| AllowAll[Allow All Endpoints]
(3) ▶ Example: Protected Product Routing
PYTHON
from fastapi import APIRouter, Depends, HTTPException, Path, Query
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional
from app.core.deps import get_db, get_current_user, require_subscription
from app.schemas import ProductCreate, ProductResponse, ProductUpdate
from app.services import ProductService
router = APIRouter()
@router.get("/products", response_model=list[ProductResponse])
async def list_products(
category: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = ProductService(db)
return await service.list_products(category=category, skip=skip, limit=limit)
@router.post("/products", response_model=ProductResponse, status_code=201)
async def create_product(
product: ProductCreate,
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = ProductService(db)
return await service.create_product(product, user_id=user.id)
@router.get("/products/{product_id}", response_model=ProductResponse)
async def get_product(
product_id: int = Path(gt=0),
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = ProductService(db)
product = await service.get_product(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
Output:
TEXT
# Function defined successfully
(4) ▶ Example: Batch Import of Endpoints (Including Subscription Limits)
PYTHON
@router.post("/products/{product_id}/prices/bulk")
async def bulk_import_prices(
product_id: int = Path(gt=0),
prices: list[PriceCreate] = ...,
db: AsyncSession = Depends(get_db),
user=Depends(require_subscription("free")), # Even free users can import
):
# Free plan: max 1000 prices per import
limits = {"free": 1000, "pro": 100000, "enterprise": 1000000}
max_import = limits.get(user.subscription, 1000)
if len(prices) > max_import:
raise HTTPException(
status_code=403,
detail=f"Import limit: {max_import} for {user.subscription} plan. Upgrade at /pricing",
)
service = PriceService(db)
return await service.bulk_import_prices(product_id, prices)
Output:
TEXT
# Function defined successfully
❓ FAQ
Q Why is CORS registered last in the middleware sequence?
A CORS needs to add CORS headers to all responses (including 4xx errors). If CORS is not the outermost layer, the 429 response returned by the rate-limiting middleware may lack the CORS header, preventing the front end from reading the error message.
Q Are route group dependencies and endpoint dependencies executed multiple times?
A No. FastAPI caches the same dependency at the request level. The
Depends(get_current_user) for a route group and the Depends(get_current_user) for an endpoint share the same execution result.Q If
get_db is declared multiple times in a DI chain, will there be multiple connections?A No.
get_db is executed only once per request, and all dependencies share the same AsyncSession.Q How can Free users upgrade to Pro?
A Stripe payment integration (beyond the scope of this lesson). Here, we use the
User.subscription field for simulation; in a production environment, this is updated via the payment callback.Q How do I test the entire request flow?
A Replace
get_db with TestClient + dependency_overrides and test step by step: No token → 401, Free token → 200 (but with restrictions), Pro token → full permissions.Q What's next after Phase 2 is complete?
A Phase 3 introduces advanced features—WebSocket real-time push, Celery asynchronous tasks, Redis caching, file uploads, pytest testing, and OpenAPI customization.
📖 Summary
- Phase 2 integrates five major modules—middleware, DI, database, CRUD, and JWT—into a complete, fully functional service
- A three-tier DI chain (get_db → get_current_user → require_subscription) centrally manages authentication and authorization
- Routing group policies: "public" requires no authentication; "api_v1" is JWT-protected; "admin" requires JWT and Admin permissions
- CORS must be registered at the outermost level to ensure that all responses (including error responses) include CORS headers
- Subscription tiers are implemented via the DI chain: Free is limited to 1,000 imports, Pro to 100,000, and Enterprise to 1,000,000.
📝 Exercises
- Basic Problem (Difficulty ⭐): Replace the in-memory CRUD implementation from Phase 1 with an asynchronous SQLAlchemy implementation, ensuring that all endpoints inject the database session via
Depends(get_db). Hint: Repository pattern +AsyncSession - Advanced Exercise (Difficulty ⭐⭐): Implement routing groups—the "public" group (/health, /auth/login) requires no authentication, while all business endpoints under "api/v1" are protected by JWT. Use Swagger UI to test logging in and accessing protected endpoints. Hint:
APIRouter(dependencies=[Depends(get_current_user)]) - Challenge (Difficulty: ⭐⭐⭐): Implement a complete three-tier DI chain + batch import of endpoints: Pro users can import 100,000 price entries; Free users will receive a 403 error if they exceed 1,000 entries. Also, fully demonstrate the registration → login → import process in Swagger UI. Hint:
require_subscription("free")+ thelimitsdictionary
---|
