Project Development — PriceTracker: From Blueprint to Code
Developing a project is like building a house—first, lay the foundation (infrastructure); next, erect the framework (authentication + products); then, add the bricks and tiles (pricing + notifications); and finally, finish the interior (reports + error handling). If you get the order wrong, the cost of rework doubles.
1. What You'll Learn
- Modular development sequence: Infrastructure → Authentication → Products → Pricing → Notifications → Reports
- Asynchronous-First Principle: End-to-end async/await, from routing to the database
- Error Handling System: Custom exception classes + global exception handler + standardized error response format
- Code quality assurance: pre-commit hooks + mypy + ruff + pytest for continuous validation
- Alice Scenario Delivery Validation: PriceTracker can handle millions of products, thousands of concurrent users, and price updates in seconds
2. Alice's True Story
(1) Pain Point: Disorganized Development Sequence Leading to Rework
Alice first developed the price import feature, but later realized that authentication and permission checks were needed, so she went back and modified 15 endpoints. Later, they discovered that the error response formats were inconsistent (some returned {"error": "..."}, while others returned {"detail": "..."}), forcing Bob to write two separate sets of front-end parsing logic. The disorganized development sequence resulted in 30% of the time being spent on rework.
(2) Solutions for Modular Development
Determine the development sequence based on dependencies: Infrastructure (DB/Redis/Config) → Authentication (JWT/Permissions) → Products (CRUD) → Pricing (Import/Push) → Notifications → Reports. Move on to the next module only after each one is completed and has passed testing; under no circumstances should work be redone.
(3) Revenue
The development rework rate dropped from 30% to 5%, and the error response format was standardized (all errors now return {"error": {"code": "...", "message": "..."}}), so Bob's front end only needs a single set of parsing logic.
3. Module Development Sequence
(1) Dependency Graph
graph TD
Infra[Infrastructure: DB/Redis/Config] --> Auth[Auth: JWT/Permissions]
Auth --> Products[Products: CRUD]
Products --> Prices[Prices: Import/Push]
Auth --> Subscriptions[Subscriptions: Plans]
Prices --> Notifications[Notifications: Alerts]
Subscriptions --> Notifications
Products --> Reports[Reports: Analytics]
Prices --> Reports
| Order | Module | Dependencies | Estimated Work Hours |
|---|---|---|---|
| 1 | Infrastructure | None | 1 day |
| 2 | Certification | Infrastructure | 1 day |
| 3 | Product CRUD | Authentication | 1 day |
| 4 | Price Import/Query | Product | 1 day |
| 5 | Subscription Plan | Certification | 0.5 days |
| 6 | Notifications & Alerts | Pricing + Subscription | 1 day |
| 7 | Report Statistics | Products + Prices | 0.5 days |
4. Error Handling System
(1) Custom Exception Classes
(1) ▶ Example: PriceTracker Exception System
PYTHON
# app/core/exceptions.py
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
class PriceTrackerError(Exception):
"""Base exception for PriceTracker"""
def __init__(self, code: str, message: str, status_code: int = 500):
self.code = code
self.message = message
self.status_code = status_code
class NotFoundError(PriceTrackerError):
def __init__(self, resource: str, resource_id: int | str):
super().__init__(
code=f"{resource.upper()}_NOT_FOUND",
message=f"{resource} with id '{resource_id}' not found",
status_code=404,
)
class SubscriptionRequiredError(PriceTrackerError):
def __init__(self, required_plan: str, current_plan: str):
super().__init__(
code="SUBSCRIPTION_REQUIRED",
message=f"This feature requires {required_plan} plan. Current: {current_plan}",
status_code=403,
)
class ImportLimitError(PriceTrackerError):
def __init__(self, limit: int, plan: str):
super().__init__(
code="IMPORT_LIMIT_EXCEEDED",
message=f"Import limit: {limit} records for {plan} plan",
status_code=403,
)
Output:
TEXT
# Function defined successfully
(2) Global Exception Handler
flowchart TD
A[Exception Raised] --> B{Exception Type?}
B -->|PriceTrackerError| C[Format standard response]
B -->|RequestValidationError| D[Format validation response]
B -->|HTTPException| E[Format HTTP response]
B -->|Other| F[Format 500 response]
C --> G[JSONResponse: code + message]
D --> G
E --> G
F --> G
(2) ▶ Example: Registering a Global Exception Handler
PYTHON
# app/core/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from app.core.exceptions import PriceTrackerError
def format_error_response(code: str, message: str, status_code: int, details=None):
return JSONResponse(
status_code=status_code,
content={
"error": {
"code": code,
"message": message,
"details": details,
}
},
)
async def pricetracker_error_handler(request: Request, exc: PriceTrackerError):
return format_error_response(exc.code, exc.message, exc.status_code)
async def validation_error_handler(request: Request, exc: RequestValidationError):
return format_error_response(
code="VALIDATION_ERROR",
message="Request validation failed",
status_code=422,
details=exc.errors(),
)
async def http_error_handler(request: Request, exc: HTTPException):
return format_error_response(
code="HTTP_ERROR",
message=str(exc.detail),
status_code=exc.status_code,
)
# Register handlers
def register_exception_handlers(app: FastAPI):
app.add_exception_handler(PriceTrackerError, pricetracker_error_handler)
app.add_exception_handler(RequestValidationError, validation_error_handler)
app.add_exception_handler(HTTPException, http_error_handler)
Output:
TEXT
# Function defined successfully
5. The Principle of Asynchronous First
(1) Full-Path Async Checklist
| Level | Synchronous ❌ | Asynchronous ✅ |
|---|---|---|
| Route | def get_xxx() |
async def get_xxx() |
| Database | Session + session.execute() |
AsyncSession + await session.execute() |
| HTTP Client | requests.get() |
httpx.AsyncClient().get() |
| Redis | redis.Redis() |
redis.asyncio.Redis() |
| File I/O | open() + read() |
aiofiles.open() + await read() |
| Task | Execute Immediately | Celery Asynchronous Task |
(1) ▶ Example: Fully Asynchronous Endpoint Implementation
PYTHON
# All layers are async
@app.get("/api/v1/products/{product_id}", response_model=ProductDetailResponse)
async def get_product_detail(
product_id: int = Path(gt=0),
db: AsyncSession = Depends(get_db), # Async DB
redis: Redis = Depends(get_redis), # Async Redis
user: User = Depends(get_current_user), # Async auth
):
# 1. Check cache (async)
cached = await redis.get(f"product:{product_id}")
if cached:
return json.loads(cached)
# 2. Query DB with eager loading (async)
stmt = (
select(Product)
.options(selectinload(Product.prices))
.where(Product.id == product_id)
)
result = await db.execute(stmt)
product = result.scalar_one_or_none()
if not product:
raise NotFoundError("product", product_id)
# 3. Cache result (async)
data = ProductDetailResponse.model_validate(product).model_dump()
await redis.setex(f"product:{product_id}", 300, json.dumps(data))
return data
Output:
TEXT
# Function defined successfully
6. Code Quality Assurance
(1) Toolchain Configuration
(1) ▶ Example: pyproject.toml Quality Tool Configuration
TOML
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM"]
[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
Output:
TEXT
CI/CD pipeline configuration has been loaded
Pipeline status: passed
Tests: 12 passed, 0 failed
(2) ▶ Example: pre-commit configuration
YAML
# .pre-commit-config.yaml
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.5.0
hooks:
- id: ruff
args: [--fix]
- id: ruff-format
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.10.0
hooks:
- id: mypy
additional_dependencies: [pydantic, sqlalchemy, fastapi]
Output:
TEXT
CI/CD pipeline loaded
Pipeline status: passed
Tests: 12 passed, 0 failed
(2) Comparison of Quality Tools
| Tool | Purpose | When to Run |
|---|---|---|
| ruff | lint + format | pre-commit + CI |
| mypy | type checking | pre-commit + CI |
| pytest | Testing | CI + Manual |
| safety | Dependency Security Scan | CI |
| bandit | Code Security Check | CI |
❓ FAQ
Q Is the development order really that important?
A It's very important. Develop the dependent modules first; modules developed later can directly use the completed functionality, thereby avoiding rework. Infrastructure → Authentication → Business Logic is a golden rule.
Q How do I choose between custom exceptions and HTTPException?
A Use custom exceptions (including code and message) for business errors, and use HTTPException for HTTP-level errors. Custom exceptions are uniformly formatted using a global handler.
Q Isn't pre-commit too slow?
A Ruff is extremely fast (Rust implementation), while Mypy is slightly slower but performs incremental checks quickly. The total time is less than 5 seconds, and in exchange, you get guaranteed code quality with every commit—which is far more efficient than having to fix issues after a CI failure.
Q Is mypy's strict mode worth it?
A Yes, it is. Strict mode catches more type errors; although the initial setup effort is high, it reduces runtime bugs in the long run. Type hints in FastAPI + Pydantic are naturally well-suited for strict mode.
Q How do you verify "one million items + thousands of concurrent requests"?
A Run load tests using Locust or k6. First, prepare one million test records (
INSERT INTO products SELECT generate_series(1, 1000000), ...), then simulate thousands of concurrent queries.Q How should the code review process be designed?
A Bob/Alice submits a PR → CI automatically runs linting and tests → Charlie reviews the code → Merge to
develop → Automatically deploy to staging → After verification, merge to main.📖 Summary
- Develop modules in the following order based on dependencies: Infrastructure → Authentication → Products → Pricing → Notifications → Reports, to avoid rework
- Custom exception system + global handler to implement a unified error response format:
{"error": {"code": "...", "message": "..."}} - Asynchronous-First Principle: Use
async/awaitthroughout the entire chain; wrap synchronous code inrun_in_executor - Code Quality Assurance: ruff (lint + format) + mypy (type checking) + pytest (testing) + pre-commit (automated validation)
- PriceTracker Delivery Validation: Millions of products, thousands of concurrent requests, push notifications within seconds, and a standardized error format
📝 Exercises
- Basic Exercise (Difficulty ⭐): Implement a custom exception hierarchy for PriceTracker (PriceTrackerError → NotFoundError → SubscriptionRequiredError), and replace
raise HTTPException(404)withraise NotFoundError("product", 42)in the endpoint. Hint: Inherit from Exception - Advanced Problem (Difficulty ⭐⭐): Implement a global exception handler and register three handlers—PriceTrackerError, RequestValidationError, and HTTPException—to ensure that all error responses follow a consistent format. Hint:
app.add_exception_handler()+format_error_response() - Challenge (Difficulty: ⭐⭐⭐): Set up a complete code quality toolchain—pyproject.toml (with ruff, mypy, and pytest configurations), .pre-commit-config.yaml—and run
ruff check+mypy app/+pytestso that all tests pass. Hint:uv add --dev ruff mypy pytest+pre-commit install
---|
