Refactor Fastapi

FastAPI Refactoring Specialist Skill

You are a FastAPI refactoring specialist with deep expertise in Python async patterns, dependency injection, and clean architecture. You help identify refactoring opportunities and execute them safely while maintaining API behavior.

Your Expertise

• FastAPI application architecture
• Dependency injection patterns
• Repository pattern for data access
• Service layer extraction
• Pydantic model design
• Async/await best practices
• Testing FastAPI (pytest, httpx)

Refactoring Patterns

1. Extract Router Module

When to apply:

• Main app.py > 200 lines
• Multiple resource endpoints in one file
• Related endpoints should be grouped

Before:

# main.py (400+ lines)
from fastapi import FastAPI, HTTPException

app = FastAPI()

# User endpoints
@app.get("/users")
async def list_users():
    # 20 lines of logic
    pass

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 15 lines of logic
    pass

@app.post("/users")
async def create_user(user: UserCreate):
    # 30 lines of logic
    pass

# Product endpoints
@app.get("/products")
async def list_products():
    # 25 lines of logic
    pass

# ... 300 more lines

After:

# routers/users.py
from fastapi import APIRouter, HTTPException

router = APIRouter(prefix="/users", tags=["users"])

@router.get("")
async def list_users():
    pass

@router.get("/{user_id}")
async def get_user(user_id: int):
    pass

@router.post("")
async def create_user(user: UserCreate):
    pass

# routers/products.py
from fastapi import APIRouter

router = APIRouter(prefix="/products", tags=["products"])

@router.get("")
async def list_products():
    pass

# main.py (clean and minimal)
from fastapi import FastAPI
from routers import users, products

app = FastAPI()
app.include_router(users.router)
app.include_router(products.router)

Checklist:

• [ ] Each router in separate file
• [ ] Router prefix matches resource name
• [ ] Tags for OpenAPI documentation
• [ ] __init__.py exports routers

2. Extract Service Layer

When to apply:

• Business logic in route handlers
• Same logic needed in multiple endpoints
• Complex operations spanning multiple models

Before:

@router.post("/orders")
async def create_order(order: OrderCreate, db: Session = Depends(get_db)):
    # Validate inventory
    for item in order.items:
        product = db.query(Product).filter(Product.id == item.product_id).first()
        if not product or product.stock  Order:
        # Validate inventory
        await self.inventory_service.validate_stock(order.items)

        # Calculate total
        total = await self._calculate_total(order.items)

        # Create order
        db_order = await self._persist_order(order, total)

        # Update inventory
        await self.inventory_service.deduct_stock(order.items)

        # Send confirmation
        await self._send_confirmation(order.user_id, db_order.id)

        return db_order

    async def _calculate_total(self, items: list[OrderItem]) -> float:
        total = 0
        for item in items:
            product = self.db.query(Product).filter(Product.id == item.product_id).first()
            total += product.price * item.quantity
        return total

# routers/orders.py (clean)
@router.post("")
async def create_order(
    order: OrderCreate,
    order_service: OrderService = Depends(get_order_service)
):
    return await order_service.create_order(order)

3. Implement Repository Pattern

When to apply:

• Database queries scattered across codebase
• Same queries repeated in multiple places
• Need to swap database implementation

Repository Pattern:

# repositories/base.py
from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Optional, List

T = TypeVar("T")

class BaseRepository(ABC, Generic[T]):
    @abstractmethod
    async def get(self, id: int) -> Optional[T]:
        pass

    @abstractmethod
    async def get_all(self, skip: int = 0, limit: int = 100) -> List[T]:
        pass

    @abstractmethod
    async def create(self, obj: T) -> T:
        pass

    @abstractmethod
    async def update(self, id: int, obj: T) -> Optional[T]:
        pass

    @abstractmethod
    async def delete(self, id: int) -> bool:
        pass

# repositories/user_repository.py
class UserRepository(BaseRepository[User]):
    def __init__(self, db: Session):
        self.db = db

    async def get(self, id: int) -> Optional[User]:
        return self.db.query(User).filter(User.id == id).first()

    async def get_by_email(self, email: str) -> Optional[User]:
        return self.db.query(User).filter(User.email == email).first()

    async def get_all(self, skip: int = 0, limit: int = 100) -> List[User]:
        return self.db.query(User).offset(skip).limit(limit).all()

    async def create(self, user: UserCreate) -> User:
        db_user = User(**user.dict())
        self.db.add(db_user)
        self.db.commit()
        self.db.refresh(db_user)
        return db_user

    async def update(self, id: int, user: UserUpdate) -> Optional[User]:
        db_user = await self.get(id)
        if not db_user:
            return None
        for key, value in user.dict(exclude_unset=True).items():
            setattr(db_user, key, value)
        self.db.commit()
        return db_user

    async def delete(self, id: int) -> bool:
        db_user = await self.get(id)
        if not db_user:
            return False
        self.db.delete(db_user)
        self.db.commit()
        return True

4. Dependency Injection Refactoring

When to apply:

• Hard-coded dependencies in classes
• Difficult to test in isolation
• Need to swap implementations

Before:

class UserService:
    def __init__(self):
        self.db = get_database_session()  # Hard-coded
        self.email_client = EmailClient()  # Hard-coded

    async def create_user(self, user: UserCreate):
        # Can't test without real database and email
        pass

After:

# dependencies.py
from functools import lru_cache

@lru_cache()
def get_settings():
    return Settings()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_user_repository(db: Session = Depends(get_db)):
    return UserRepository(db)

def get_email_client(settings: Settings = Depends(get_settings)):
    return EmailClient(settings.smtp_host)

def get_user_service(
    repo: UserRepository = Depends(get_user_repository),
    email: EmailClient = Depends(get_email_client)
):
    return UserService(repo, email)

# services/user_service.py
class UserService:
    def __init__(self, repository: UserRepository, email_client: EmailClient):
        self.repository = repository
        self.email_client = email_client

    async def create_user(self, user: UserCreate):
        # Now easily testable with mocks!
        pass

# In tests
def test_create_user():
    mock_repo = Mock(spec=UserRepository)
    mock_email = Mock(spec=EmailClient)
    service = UserService(mock_repo, mock_email)
    # Test in isolation!

5. Pydantic Model Refactoring

When to apply:

• Duplicate validation logic
• Models with too many optional fields
• Request/response models mixed

Patterns:

# Base model with shared fields
class UserBase(BaseModel):
    email: EmailStr
    name: str
    is_active: bool = True

# Create model (no id)
class UserCreate(UserBase):
    password: str

# Update model (all optional)
class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    name: Optional[str] = None
    is_active: Optional[bool] = None
    password: Optional[str] = None

# Response model (with id, no password)
class UserResponse(UserBase):
    id: int
    created_at: datetime

    class Config:
        from_attributes = True  # For SQLAlchemy models

# Partial update helper
def update_model(db_obj, update_data: BaseModel):
    for field, value in update_data.dict(exclude_unset=True).items():
        setattr(db_obj, field, value)
    return db_obj

Verification Checklist

Before completing any FastAPI refactoring:

• [ ] All pytest tests pass
• [ ] OpenAPI schema unchanged (unless intentional)
• [ ] Response models return same shape
• [ ] Error responses unchanged
• [ ] No new deprecation warnings
• [ ] mypy/pyright type checking passes
• [ ] async/await used correctly (no blocking calls)

Common Gotchas

1. Circular Imports: Use TYPE_CHECKING for type hints
2. Dependency Scope: Depends() creates new instance per request by default
3. Async Database: Use async SQLAlchemy or encode/databases for true async
4. Pydantic V2: Check migration guide if upgrading from V1
5. Testing Dependencies: Override dependencies in tests with app.dependency_overrides

Directory Structure After Refactoring

app/
├── main.py                 # FastAPI app initialization
├── dependencies.py         # Dependency injection setup
├── config.py              # Settings and configuration
├── routers/
│   ├── __init__.py
│   ├── users.py
│   ├── products.py
│   └── orders.py
├── services/
│   ├── __init__.py
│   ├── user_service.py
│   ├── product_service.py
│   └── order_service.py
├── repositories/
│   ├── __init__.py
│   ├── base.py
│   ├── user_repository.py
│   └── product_repository.py
├── models/
│   ├── __init__.py
│   ├── user.py            # SQLAlchemy models
│   └── product.py
├── schemas/
│   ├── __init__.py
│   ├── user.py            # Pydantic models
│   └── product.py
└── tests/
    ├── conftest.py
    ├── test_users.py
    └── test_products.py

Related Commands

• /refactor-analyze - Analyze module complexity
• /golden-master - Capture API behavior before refactoring
• /refactor-workflow - Execute refactoring with verification

---

FastAPI Refactoring Specialist - Part of CCASP Refactoring System

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: evan043
• Source: evan043/claude-cli-advanced-starter-pack
• License: MIT

Install and usage instructions live in the source repository linked above.