# Copy the blueprint to your new project
cp -r FastAPIBluePrint my-new-project
cd my-new-project
# Create and activate a virtual environment
python3.10 -m venv venv
source venv/bin/activate
# Install dependencies
pip install -r requirements.txt
# Copy environment template
cp .env.example .envBefore making changes, read:
- Architecture Overview
- Project Structure
- Review the skeleton code in
src/
# Development mode (with auto-reload)
uvicorn src.main:app --reload
# Production mode
uvicorn src.main:appVisit http://localhost:8000/docs for interactive API documentation.
# Run all tests
pytest
# Run with coverage
pytest --cov=src
# Run specific test file
pytest tests/unit/application/test_example_use_case.py
# Run with verbose output
pytest -vThis section walks through adding a complete feature from domain to API.
File: src/domain/entities/product.py
"""Product entity - core domain model."""
from datetime import datetime
from decimal import Decimal
class Product:
"""
Product entity representing a physical or digital product.
Encapsulates product-related business rules and invariants.
Note: This entity is framework-agnostic and has NO external dependencies.
"""
def __init__(
self,
id: int | None,
name: str,
description: str,
price: Decimal,
created_at: datetime | None = None,
updated_at: datetime | None = None,
):
"""
Initialize a Product entity.
Args:
id: Unique identifier (None for new products)
name: Product name
description: Product description
price: Product price in decimal format
created_at: Creation timestamp
updated_at: Last update timestamp
Raises:
ValueError: If price is negative or name is empty
"""
self.id = id
self.name = name
self.description = description
self.price = price
self.created_at = created_at or datetime.utcnow()
self.updated_at = updated_at or datetime.utcnow()
# Validate business invariants
self._validate()
def _validate(self) -> None:
"""Enforce business rules and invariants."""
if not self.name or len(self.name.strip()) == 0:
raise ValueError("Product name cannot be empty")
if self.price < 0:
raise ValueError("Product price cannot be negative")
def update_price(self, new_price: Decimal) -> None:
"""
Update product price.
Business rule: Price cannot be negative.
"""
if new_price < 0:
raise ValueError("Product price cannot be negative")
self.price = new_price
self.updated_at = datetime.utcnow()
def deactivate(self) -> None:
"""
Mark product as inactive.
Business rule: Deactivated products cannot be reactivated.
"""
self.is_active = False
self.updated_at = datetime.utcnow()
def __eq__(self, other: object) -> bool:
"""Products are equal if they have the same ID."""
if not isinstance(other, Product):
return NotImplemented
return self.id == other.id and self.id is not None
def __hash__(self) -> int:
"""Products are hashable by ID."""
return hash(self.id) if self.id is not None else hash(id(self))
def __repr__(self) -> str:
return f"Product(id={self.id}, name={self.name!r}, price={self.price})"File: src/domain/repositories/product_repository.py
"""Product repository interface - data access contract."""
from abc import ABC, abstractmethod
from decimal import Decimal
from src.domain.entities.product import Product
class IProductRepository(ABC):
"""
Interface for product persistence.
This interface defines WHAT data operations are needed,
not HOW they are implemented. Implementations can use:
- SQLAlchemy (SQL databases)
- MongoDB (document databases)
- Redis (in-memory cache)
- Any other persistence mechanism
Use cases depend on this interface, not concrete implementations.
This enables testing with mock repositories and easy database switching.
"""
@abstractmethod
async def find_by_id(self, product_id: int) -> Product | None:
"""
Find a product by ID.
Args:
product_id: Product identifier
Returns:
Product entity if found, None otherwise
Raises:
RepositoryError: If database error occurs
"""
pass
@abstractmethod
async def find_by_name(self, name: str) -> Product | None:
"""
Find a product by name.
Args:
name: Product name
Returns:
Product entity if found, None otherwise
"""
pass
@abstractmethod
async def find_all(
self,
skip: int = 0,
limit: int = 10,
) -> list[Product]:
"""
Find all products with pagination.
Args:
skip: Number of records to skip
limit: Maximum number of records to return
Returns:
List of product entities
"""
pass
@abstractmethod
async def save(self, product: Product) -> Product:
"""
Persist a product (insert or update).
Args:
product: Product entity to save
Returns:
Persisted product entity (with ID if newly created)
Raises:
RepositoryError: If database error occurs
UniqueConstraintError: If name is not unique
"""
pass
@abstractmethod
async def delete(self, product_id: int) -> bool:
"""
Delete a product by ID.
Args:
product_id: Product identifier
Returns:
True if deleted, False if not found
Raises:
RepositoryError: If database error occurs
"""
pass
@abstractmethod
async def count(self) -> int:
"""
Count total number of products.
Returns:
Total product count
"""
passFile: src/domain/exceptions/product_exceptions.py
"""Product domain exceptions."""
class ProductError(Exception):
"""Base exception for product domain errors."""
pass
class ProductNotFoundError(ProductError):
"""Raised when a product is not found."""
def __init__(self, product_id: int):
self.product_id = product_id
super().__init__(f"Product with ID {product_id} not found")
class ProductAlreadyExistsError(ProductError):
"""Raised when trying to create a product that already exists."""
def __init__(self, name: str):
self.name = name
super().__init__(f"Product with name '{name}' already exists")
class InvalidProductDataError(ProductError):
"""Raised when product data is invalid."""
pass
class ProductPriceMustBePositiveError(ProductError):
"""Raised when product price is negative or zero."""
def __init__(self, price: float):
self.price = price
super().__init__(f"Product price must be positive, got {price}")File: src/application/dto/product_dto.py
"""Product Data Transfer Objects (DTOs)."""
from datetime import datetime
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, Field
class CreateProductInputDTO(BaseModel):
"""DTO for creating a new product."""
name: str = Field(..., min_length=1, max_length=200)
description: str = Field(..., min_length=1, max_length=2000)
price: Decimal = Field(..., gt=0, decimal_places=2)
model_config = {"from_attributes": True}
class UpdateProductInputDTO(BaseModel):
"""DTO for updating an existing product."""
name: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = Field(None, min_length=1, max_length=2000)
price: Optional[Decimal] = Field(None, gt=0, decimal_places=2)
model_config = {"from_attributes": True}
class ProductOutputDTO(BaseModel):
"""DTO returned from product operations."""
id: int
name: str
description: str
price: Decimal
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
class ProductListOutputDTO(BaseModel):
"""DTO for listing products with pagination."""
items: list[ProductOutputDTO]
total: int
skip: int
limit: intFile: src/application/mappers/product_mapper.py
"""Product mapper - converts between domain entities and DTOs."""
from src.application.dto.product_dto import (
ProductOutputDTO,
ProductListOutputDTO,
)
from src.domain.entities.product import Product
class ProductMapper:
"""
Maps between Product entities and DTOs.
Responsibility: Ensure DTOs don't leak domain details.
Translates domain concepts to API representations.
"""
@staticmethod
def to_output_dto(product: Product) -> ProductOutputDTO:
"""
Convert domain Product entity to output DTO.
Args:
product: Domain entity
Returns:
Output DTO ready for API response
"""
return ProductOutputDTO(
id=product.id,
name=product.name,
description=product.description,
price=product.price,
created_at=product.created_at,
updated_at=product.updated_at,
)
@staticmethod
def to_output_dtos(products: list[Product]) -> list[ProductOutputDTO]:
"""Convert multiple entities to DTOs."""
return [ProductMapper.to_output_dto(p) for p in products]
@staticmethod
def to_list_output_dto(
products: list[Product],
total: int,
skip: int,
limit: int,
) -> ProductListOutputDTO:
"""Convert paginated products to list DTO."""
return ProductListOutputDTO(
items=ProductMapper.to_output_dtos(products),
total=total,
skip=skip,
limit=limit,
)File: src/application/use_cases/product_use_cases.py
"""Product use cases - application service layer."""
from decimal import Decimal
from src.application.dto.product_dto import (
CreateProductInputDTO,
ProductOutputDTO,
ProductListOutputDTO,
)
from src.application.mappers.product_mapper import ProductMapper
from src.domain.entities.product import Product
from src.domain.exceptions.product_exceptions import (
ProductNotFoundError,
ProductAlreadyExistsError,
)
from src.domain.repositories.product_repository import IProductRepository
class CreateProductUseCase:
"""
Use case for creating a new product.
Responsibility:
- Enforce business rules (e.g., unique product name)
- Orchestrate domain objects
- Persist through repository interface
"""
def __init__(self, product_repository: IProductRepository):
self.product_repository = product_repository
async def execute(self, input_dto: CreateProductInputDTO) -> ProductOutputDTO:
"""
Create a new product.
Business rules:
- Product name must be unique
- Price must be positive
Args:
input_dto: Product creation input
Returns:
Created product as DTO
Raises:
ProductAlreadyExistsError: If product name exists
"""
# Check business rule: unique product name
existing = await self.product_repository.find_by_name(input_dto.name)
if existing:
raise ProductAlreadyExistsError(input_dto.name)
# Create domain entity
product = Product(
id=None,
name=input_dto.name,
description=input_dto.description,
price=input_dto.price,
)
# Persist through repository interface
created_product = await self.product_repository.save(product)
# Return as DTO
return ProductMapper.to_output_dto(created_product)
class GetProductUseCase:
"""Use case for retrieving a product by ID."""
def __init__(self, product_repository: IProductRepository):
self.product_repository = product_repository
async def execute(self, product_id: int) -> ProductOutputDTO:
"""
Get a product by ID.
Args:
product_id: Product identifier
Returns:
Product as DTO
Raises:
ProductNotFoundError: If product doesn't exist
"""
product = await self.product_repository.find_by_id(product_id)
if not product:
raise ProductNotFoundError(product_id)
return ProductMapper.to_output_dto(product)
class ListProductsUseCase:
"""Use case for listing all products with pagination."""
def __init__(self, product_repository: IProductRepository):
self.product_repository = product_repository
async def execute(
self,
skip: int = 0,
limit: int = 10,
) -> ProductListOutputDTO:
"""
List products with pagination.
Args:
skip: Number of records to skip
limit: Maximum records to return
Returns:
Paginated product list as DTO
"""
products = await self.product_repository.find_all(skip=skip, limit=limit)
total = await self.product_repository.count()
return ProductMapper.to_list_output_dto(products, total, skip, limit)
class UpdateProductPriceUseCase:
"""Use case for updating product price."""
def __init__(self, product_repository: IProductRepository):
self.product_repository = product_repository
async def execute(
self,
product_id: int,
new_price: Decimal,
) -> ProductOutputDTO:
"""
Update product price.
Args:
product_id: Product identifier
new_price: New price
Returns:
Updated product as DTO
Raises:
ProductNotFoundError: If product doesn't exist
"""
product = await self.product_repository.find_by_id(product_id)
if not product:
raise ProductNotFoundError(product_id)
# Business logic: update price through entity method
product.update_price(new_price)
# Persist changes
updated = await self.product_repository.save(product)
return ProductMapper.to_output_dto(updated)
class DeleteProductUseCase:
"""Use case for deleting a product."""
def __init__(self, product_repository: IProductRepository):
self.product_repository = product_repository
async def execute(self, product_id: int) -> bool:
"""
Delete a product.
Args:
product_id: Product identifier
Returns:
True if deleted, False if not found
"""
return await self.product_repository.delete(product_id)File: src/infrastructure/database/models.py (add Product model)
"""SQLAlchemy ORM models - database representation."""
from datetime import datetime
from decimal import Decimal
from sqlalchemy import Column, Integer, String, Numeric, DateTime
from src.infrastructure.database.base import Base
class ProductModel(Base):
"""
SQLAlchemy ORM model for Product persistence.
Note: This is an infrastructure detail, separate from domain entities.
The domain Product entity doesn't know about this model.
"""
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(200), unique=True, index=True, nullable=False)
description = Column(String(2000), nullable=False)
price = Column(Numeric(precision=19, scale=2), nullable=False)
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
def __repr__(self) -> str:
return f"<ProductModel(id={self.id}, name={self.name!r})>"File: src/infrastructure/repositories/sqlalchemy_product_repository.py
"""Product repository implementation using SQLAlchemy."""
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.domain.entities.product import Product
from src.domain.repositories.product_repository import IProductRepository
from src.infrastructure.database.models import ProductModel
class SQLAlchemyProductRepository(IProductRepository):
"""
Concrete product repository using SQLAlchemy ORM.
Responsibility:
- Translate between ORM models and domain entities
- Execute database queries
- Handle persistence operations
If you need to switch to MongoDB or another database,
create a new implementation (e.g., MongoDBProductRepository)
without changing use cases.
"""
def __init__(self, session: AsyncSession):
"""
Initialize repository with database session.
Args:
session: SQLAlchemy AsyncSession for database operations
"""
self.session = session
async def find_by_id(self, product_id: int) -> Product | None:
"""Find product by ID."""
query = select(ProductModel).where(ProductModel.id == product_id)
result = await self.session.execute(query)
model = result.scalar_one_or_none()
return self._to_domain(model) if model else None
async def find_by_name(self, name: str) -> Product | None:
"""Find product by name."""
query = select(ProductModel).where(ProductModel.name == name)
result = await self.session.execute(query)
model = result.scalar_one_or_none()
return self._to_domain(model) if model else None
async def find_all(
self,
skip: int = 0,
limit: int = 10,
) -> list[Product]:
"""Find all products with pagination."""
query = select(ProductModel).offset(skip).limit(limit)
result = await self.session.execute(query)
models = result.scalars().all()
return [self._to_domain(m) for m in models]
async def save(self, product: Product) -> Product:
"""Save (insert or update) a product."""
model = self._to_model(product)
self.session.add(model)
await self.session.flush()
# Refresh to get generated ID if it's a new product
await self.session.refresh(model)
return self._to_domain(model)
async def delete(self, product_id: int) -> bool:
"""Delete a product by ID."""
query = select(ProductModel).where(ProductModel.id == product_id)
result = await self.session.execute(query)
model = result.scalar_one_or_none()
if not model:
return False
await self.session.delete(model)
await self.session.flush()
return True
async def count(self) -> int:
"""Count total products."""
query = select(ProductModel)
result = await self.session.execute(query)
return len(result.scalars().all())
@staticmethod
def _to_domain(model: ProductModel) -> Product:
"""
Convert ORM model to domain entity.
This ensures domain entities are decoupled from database representation.
"""
return Product(
id=model.id,
name=model.name,
description=model.description,
price=model.price,
created_at=model.created_at,
updated_at=model.updated_at,
)
@staticmethod
def _to_model(product: Product) -> ProductModel:
"""
Convert domain entity to ORM model for persistence.
"""
return ProductModel(
id=product.id,
name=product.name,
description=product.description,
price=product.price,
created_at=product.created_at,
updated_at=product.updated_at,
)File: src/presentation/schemas/product_schemas.py
"""Pydantic schemas for product API requests and responses."""
from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field
from src.application.dto.product_dto import ProductOutputDTO
class CreateProductRequest(BaseModel):
"""
Request schema for creating a product.
This validates the incoming HTTP request data.
FastAPI uses this for:
- Request validation
- API documentation (OpenAPI/Swagger)
- Automatic error responses for invalid data
"""
name: str = Field(
...,
min_length=1,
max_length=200,
description="Product name"
)
description: str = Field(
...,
min_length=1,
max_length=2000,
description="Product description"
)
price: Decimal = Field(
...,
gt=Decimal("0"),
decimal_places=2,
description="Product price"
)
class UpdateProductRequest(BaseModel):
"""Request schema for updating a product."""
name: str | None = Field(
None,
min_length=1,
max_length=200,
description="Product name"
)
description: str | None = Field(
None,
min_length=1,
max_length=2000,
description="Product description"
)
price: Decimal | None = Field(
None,
gt=Decimal("0"),
decimal_places=2,
description="Product price"
)
class ProductResponse(BaseModel):
"""
Response schema for product data.
This is returned in API responses and documents the response structure.
"""
id: int
name: str
description: str
price: Decimal
created_at: datetime
updated_at: datetime
@classmethod
def from_dto(cls, dto: ProductOutputDTO) -> "ProductResponse":
"""Create response from DTO."""
return cls(**dto.model_dump())
class ProductListResponse(BaseModel):
"""Response schema for paginated product list."""
items: list[ProductResponse]
total: int
skip: int
limit: intFile: src/presentation/api/product_router.py
"""Product API router - FastAPI endpoints."""
from decimal import Decimal
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from src.application.use_cases.product_use_cases import (
CreateProductUseCase,
GetProductUseCase,
ListProductsUseCase,
UpdateProductPriceUseCase,
DeleteProductUseCase,
)
from src.domain.exceptions.product_exceptions import (
ProductNotFoundError,
ProductAlreadyExistsError,
)
from src.infrastructure.repositories.sqlalchemy_product_repository import (
SQLAlchemyProductRepository,
)
from src.presentation.api.dependencies import get_db_session
from src.presentation.schemas.product_schemas import (
CreateProductRequest,
UpdateProductRequest,
ProductResponse,
ProductListResponse,
)
# Router instance with prefix and tags for OpenAPI
router = APIRouter(prefix="/api/products", tags=["products"])
# ============================================================================
# Dependency Injection: Create use case dependencies
# ============================================================================
def get_product_repository(
session: AsyncSession = Depends(get_db_session),
) -> SQLAlchemyProductRepository:
"""Provide product repository with database session."""
return SQLAlchemyProductRepository(session)
def get_create_product_use_case(
repository: SQLAlchemyProductRepository = Depends(get_product_repository),
) -> CreateProductUseCase:
"""Provide CreateProductUseCase with repository."""
return CreateProductUseCase(repository)
def get_get_product_use_case(
repository: SQLAlchemyProductRepository = Depends(get_product_repository),
) -> GetProductUseCase:
"""Provide GetProductUseCase with repository."""
return GetProductUseCase(repository)
def get_list_products_use_case(
repository: SQLAlchemyProductRepository = Depends(get_product_repository),
) -> ListProductsUseCase:
"""Provide ListProductsUseCase with repository."""
return ListProductsUseCase(repository)
def get_update_product_price_use_case(
repository: SQLAlchemyProductRepository = Depends(get_product_repository),
) -> UpdateProductPriceUseCase:
"""Provide UpdateProductPriceUseCase with repository."""
return UpdateProductPriceUseCase(repository)
def get_delete_product_use_case(
repository: SQLAlchemyProductRepository = Depends(get_product_repository),
) -> DeleteProductUseCase:
"""Provide DeleteProductUseCase with repository."""
return DeleteProductUseCase(repository)
# ============================================================================
# API Endpoints
# ============================================================================
@router.post("/", response_model=ProductResponse, status_code=201)
async def create_product(
request: CreateProductRequest,
use_case: CreateProductUseCase = Depends(get_create_product_use_case),
) -> ProductResponse:
"""
Create a new product.
- **name**: Product name (required, 1-200 characters)
- **description**: Product description (required, 1-2000 characters)
- **price**: Product price in USD (required, must be positive)
Returns:
- Created product with ID and timestamps
Status Codes:
- 201: Product created successfully
- 409: Product with this name already exists
- 422: Invalid request data
"""
try:
from src.application.dto.product_dto import CreateProductInputDTO
input_dto = CreateProductInputDTO(
name=request.name,
description=request.description,
price=request.price,
)
result = await use_case.execute(input_dto)
return ProductResponse.from_dto(result)
except ProductAlreadyExistsError as e:
raise HTTPException(
status_code=409,
detail=f"Product with name '{request.name}' already exists"
)
except ValueError as e:
raise HTTPException(
status_code=422,
detail=str(e)
)
@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(
product_id: int,
use_case: GetProductUseCase = Depends(get_get_product_use_case),
) -> ProductResponse:
"""
Retrieve a product by ID.
Args:
- **product_id**: Product identifier (path parameter)
Returns:
- Product details
Status Codes:
- 200: Product found
- 404: Product not found
"""
try:
result = await use_case.execute(product_id)
return ProductResponse.from_dto(result)
except ProductNotFoundError:
raise HTTPException(
status_code=404,
detail=f"Product with ID {product_id} not found"
)
@router.get("/", response_model=ProductListResponse)
async def list_products(
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(10, ge=1, le=100, description="Maximum records to return"),
use_case: ListProductsUseCase = Depends(get_list_products_use_case),
) -> ProductListResponse:
"""
List all products with pagination.
Query Parameters:
- **skip**: Number of records to skip (default: 0)
- **limit**: Maximum records to return (default: 10, max: 100)
Returns:
- Paginated list of products with total count
Status Codes:
- 200: Success
"""
result = await use_case.execute(skip=skip, limit=limit)
return ProductListResponse(
items=[ProductResponse.from_dto(item) for item in result.items],
total=result.total,
skip=result.skip,
limit=result.limit,
)
@router.patch("/{product_id}/price", response_model=ProductResponse)
async def update_product_price(
product_id: int,
new_price: Decimal = Query(..., gt=0, description="New price"),
use_case: UpdateProductPriceUseCase = Depends(get_update_product_price_use_case),
) -> ProductResponse:
"""
Update a product's price.
Args:
- **product_id**: Product identifier (path parameter)
- **new_price**: New price (query parameter)
Returns:
- Updated product
Status Codes:
- 200: Price updated
- 404: Product not found
"""
try:
result = await use_case.execute(product_id, new_price)
return ProductResponse.from_dto(result)
except ProductNotFoundError:
raise HTTPException(
status_code=404,
detail=f"Product with ID {product_id} not found"
)
@router.delete("/{product_id}", status_code=204)
async def delete_product(
product_id: int,
use_case: DeleteProductUseCase = Depends(get_delete_product_use_case),
) -> None:
"""
Delete a product.
Args:
- **product_id**: Product identifier
Status Codes:
- 204: Product deleted successfully
- 404: Product not found
"""
deleted = await use_case.execute(product_id)
if not deleted:
raise HTTPException(
status_code=404,
detail=f"Product with ID {product_id} not found"
)File: src/presentation/api/dependencies.py
"""Dependency injection setup for FastAPI."""
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from src.infrastructure.database.session import AsyncSessionLocal
async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
"""
Provide a database session for each request.
This dependency:
- Creates a new database session for each request
- Automatically closes the session after request completes
- Can be used in any endpoint or other dependencies
Usage in endpoints:
@router.get("/")
async def my_endpoint(session: AsyncSession = Depends(get_db_session)):
...
"""
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()File: tests/unit/application/test_product_use_cases.py
"""Unit tests for product use cases."""
import pytest
from decimal import Decimal
from unittest.mock import AsyncMock
from src.application.dto.product_dto import CreateProductInputDTO
from src.application.use_cases.product_use_cases import CreateProductUseCase
from src.domain.entities.product import Product
from src.domain.exceptions.product_exceptions import ProductAlreadyExistsError
from src.domain.repositories.product_repository import IProductRepository
class MockProductRepository(IProductRepository):
"""Mock repository for testing."""
def __init__(self):
self.products = {}
self.next_id = 1
async def find_by_id(self, product_id: int) -> Product | None:
return self.products.get(product_id)
async def find_by_name(self, name: str) -> Product | None:
for product in self.products.values():
if product.name == name:
return product
return None
async def find_all(self, skip: int = 0, limit: int = 10) -> list[Product]:
items = list(self.products.values())
return items[skip : skip + limit]
async def save(self, product: Product) -> Product:
if product.id is None:
product.id = self.next_id
self.next_id += 1
self.products[product.id] = product
return product
async def delete(self, product_id: int) -> bool:
if product_id in self.products:
del self.products[product_id]
return True
return False
async def count(self) -> int:
return len(self.products)
@pytest.mark.asyncio
class TestCreateProductUseCase:
"""Tests for CreateProductUseCase."""
async def test_create_product_success(self):
"""Test successful product creation."""
# Arrange
repository = MockProductRepository()
use_case = CreateProductUseCase(repository)
input_dto = CreateProductInputDTO(
name="Test Product",
description="A test product",
price=Decimal("99.99"),
)
# Act
result = await use_case.execute(input_dto)
# Assert
assert result.id is not None
assert result.name == "Test Product"
assert result.price == Decimal("99.99")
async def test_create_product_with_duplicate_name(self):
"""Test that duplicate product names are rejected."""
# Arrange
repository = MockProductRepository()
use_case = CreateProductUseCase(repository)
# Create first product
input_dto1 = CreateProductInputDTO(
name="Test Product",
description="First product",
price=Decimal("99.99"),
)
await use_case.execute(input_dto1)
# Try to create second product with same name
input_dto2 = CreateProductInputDTO(
name="Test Product",
description="Second product",
price=Decimal("49.99"),
)
# Act & Assert
with pytest.raises(ProductAlreadyExistsError):
await use_case.execute(input_dto2)File: tests/integration/api/test_product_endpoints.py
"""Integration tests for product API endpoints."""
import pytest
from httpx import AsyncClient
from decimal import Decimal
from src.main import app
from src.infrastructure.database.session import AsyncSessionLocal
@pytest.mark.asyncio
class TestProductEndpoints:
"""Tests for product API endpoints."""
async def test_create_product(self):
"""Test product creation endpoint."""
# Arrange
async with AsyncClient(app=app, base_url="http://test") as client:
# Act
response = await client.post(
"/api/products/",
json={
"name": "Test Product",
"description": "A test product",
"price": "99.99"
}
)
# Assert
assert response.status_code == 201
data = response.json()
assert data["name"] == "Test Product"
assert float(data["price"]) == 99.99
async def test_get_product(self):
"""Test get product endpoint."""
async with AsyncClient(app=app, base_url="http://test") as client:
# Create a product first
create_response = await client.post(
"/api/products/",
json={
"name": "Get Test Product",
"description": "For testing get",
"price": "49.99"
}
)
product_id = create_response.json()["id"]
# Get the product
response = await client.get(f"/api/products/{product_id}")
assert response.status_code == 200
assert response.json()["name"] == "Get Test Product"
async def test_list_products(self):
"""Test list products endpoint."""
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/api/products/?skip=0&limit=10")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in dataKeep it simple:
- Single file per feature in each layer
- Minimal configuration
- Inline dependencies
src/
├── domain/entities.py # All entities in one file
├── application/use_cases.py # All use cases in one file
├── infrastructure/repos.py # All repositories in one file
└── presentation/routers.py # All routers in one file
Use the blueprint structure:
- One feature per folder
- Clear separation of concerns
- Enhanced error handling
src/
├── domain/
│ ├── products/
│ │ ├── entities.py
│ │ ├── repositories.py
│ │ └── exceptions.py
│ └── users/
│ ├── entities.py
│ ├── repositories.py
│ └── exceptions.py
Add layers:
- Event-driven architecture
- Messaging/queues
- Domain events
- Bounded contexts
src/
├── shared/ # Shared across domains
├── products/ # Bounded context
│ ├── domain/
│ ├── application/
│ ├── infrastructure/
│ └── presentation/
├── users/ # Another bounded context
│ ├── domain/
│ ├── application/
│ ├── infrastructure/
│ └── presentation/
└── events/ # Event-driven integration
- Domain Layer Is Sacred: Keep it pure; no external dependencies
- Use Cases Orchestrate: They're the business logic coordinator
- Repository Pattern Rocks: Swap databases without touching business logic
- Dependency Injection: Enables testing and flexibility
- Test Your Use Cases: Most business logic lives here
- DTOs Decouple: Don't leak domain details to the API
❌ DON'T: Import FastAPI in domain or application layers
✅ DO: Keep those layers framework-agnostic
❌ DON'T: Put database queries directly in use cases
✅ DO: Use repository interfaces
❌ DON'T: Return ORM models from routers
✅ DO: Return DTOs/response schemas
❌ DON'T: Create dependencies inside use cases
✅ DO: Inject dependencies via constructor
Next Steps:
- Run
pytestto execute the example tests - Start the application with
uvicorn src.main:app --reload - Visit
http://localhost:8000/docsto explore the API - Read the Testing Strategy for detailed testing guidance