FastAPI Senior Developer

Transform into a Senior Python Backend Engineer for production-ready FastAPI applications.

When to Use

• Scaffolding new FastAPI projects
• Implementing clean architecture patterns
• Database integration (PostgreSQL, MongoDB)
• Authentication (OAuth2, JWT, OIDC)
• Microservices & event-driven patterns
• Performance optimization & async patterns
• Security hardening (OWASP compliance)

Triggers

• /fastapi-init - Scaffold new project with clean architecture
• /fastapi-structure - Analyze & restructure existing project
• /fastapi-audit - Code review for patterns, performance, security

Reference Files

Load appropriate references based on task context:

Core Tenets

1. Thin Routes, Fat Services

Routes handle HTTP concerns only. Business logic lives in services.

# WRONG: Logic in route
@router.post("/orders")
async def create_order(order: OrderCreate, db: AsyncSession = Depends(get_db)):
    if not await db.get(Product, order.product_id):
        raise HTTPException(404, "Product not found")
    # ... 50 more lines of business logic
    return order

# RIGHT: Thin route, fat service
@router.post("/orders", response_model=OrderResponse)
async def create_order(
    order: OrderCreate,
    service: OrderService = Depends(get_order_service)
) -> OrderResponse:
    return await service.create(order)

2. Configuration First

Use pydantic-settings as foundational concern. Split by domain.

# core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class DatabaseSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="DB_")

    host: str = "localhost"
    port: int = 5432
    name: str
    user: str
    password: str
    pool_size: int = 10
    max_overflow: int = 20

    @property
    def async_url(self) -> str:
        return f"postgresql+asyncpg://{self.user}:{self.password}@{self.host}:{self.port}/{self.name}"

class AuthSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="AUTH_")

    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    refresh_token_expire_days: int = 7

class Settings(BaseSettings):
    debug: bool = False
    db: DatabaseSettings = DatabaseSettings()
    auth: AuthSettings = AuthSettings()

settings = Settings()

3. Project Organization

Choose architecture based on project size. Be consistent.

Vertical Slice (Recommended for most projects)

src/
├── users/
│   ├── router.py
│   ├── service.py
│   ├── schemas.py
│   ├── models.py
│   └── dependencies.py
├── orders/
│   ├── router.py
│   ├── service.py
│   └── ...
└── core/
    ├── config.py
    ├── database.py
    └── security.py

Layered Architecture (Large teams, strict boundaries)

src/
├── api/
│   ├── routes/
│   ├── deps/
│   └── schemas/
├── services/
├── repositories/
├── models/
│   ├── domain/
│   └── db/
└── core/

4. Service Layer Pattern (Not Repository)

Use services with direct ORM access. Avoid unnecessary repository abstraction.

# services/user_service.py
class UserService:
    def __init__(self, db: AsyncSession, cache: Redis):
        self.db = db
        self.cache = cache

    async def get_by_id(self, user_id: int) -> User | None:
        # Check cache first
        cached = await self.cache.get(f"user:{user_id}")
        if cached:
            return User.model_validate_json(cached)

        # Direct ORM query - no repository needed
        result = await self.db.execute(
            select(UserModel)
            .options(selectinload(UserModel.profile))
            .where(UserModel.id == user_id)
        )
        user_model = result.scalar_one_or_none()

        if user_model:
            user = User.model_validate(user_model)
            await self.cache.setex(f"user:{user_id}", 300, user.model_dump_json())
            return user
        return None

5. Advanced Dependency Injection

Chain dependencies for validation and composition.

# deps/common.py
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# deps/users.py
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    payload = verify_token(token)
    user = await db.get(UserModel, payload["sub"])
    if not user:
        raise HTTPException(401, "User not found")
    return user

async def get_current_active_user(
    user: User = Depends(get_current_user)
) -> User:
    if not user.is_active:
        raise HTTPException(403, "Inactive user")
    return user

# deps/resources.py
async def valid_post_id(
    post_id: int,
    db: AsyncSession = Depends(get_db)
) -> Post:
    post = await db.get(PostModel, post_id)
    if not post:
        raise HTTPException(404, "Post not found")
    return post

async def valid_owned_post(
    post: Post = Depends(valid_post_id),
    user: User = Depends(get_current_user)
) -> Post:
    if post.owner_id != user.id:
        raise HTTPException(403, "Not your post")
    return post

# Usage in routes
@router.put("/posts/{post_id}")
async def update_post(
    data: PostUpdate,
    post: Post = Depends(valid_owned_post)  # Validates existence + ownership
) -> PostResponse:
    ...

Async Patterns

Do

# Async DB with proper session handling
async def get_user(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# Concurrent independent calls
async def get_dashboard_data(user_id: int) -> DashboardData:
    user, orders, notifications = await asyncio.gather(
        user_service.get(user_id),
        order_service.list_recent(user_id),
        notification_service.get_unread(user_id),
        return_exceptions=True
    )
    return DashboardData(user=user, orders=orders, notifications=notifications)

# Background tasks for non-blocking operations
@router.post("/users")
async def create_user(user: UserCreate, background: BackgroundTasks):
    db_user = await user_service.create(user)
    background.add_task(send_welcome_email, db_user.email)
    background.add_task(analytics.track, "user_created", db_user.id)
    return db_user

Don't

# WRONG: Blocking calls in async context
time.sleep(5)           # Use: await asyncio.sleep(5)
requests.get(url)       # Use: async with httpx.AsyncClient() as client
open("file").read()     # Use: aiofiles.open()

# WRONG: Sequential when parallel is possible
user = await get_user(id)
orders = await get_orders(id)  # Use asyncio.gather()

# WRONG: Sync dependencies in async routes
def get_db():  # Should be: async def get_db()
    return SessionLocal()
`