skill
---
name: FastAPI Project Structure
description: Layered architecture with service pattern and dependency injection
---
# FastAPI Project Structure
## Layered Architecture
```
┌─────────────────────────────────────────┐
│ Router Layer │ ← HTTP handling, validation
├─────────────────────────────────────────┤
│ Service Layer │ ← Business logic
├─────────────────────────────────────────┤
│ Repository Layer │ ← Data access (ORM)
├─────────────────────────────────────────┤
│ Database │ ← PostgreSQL
└─────────────────────────────────────────┘
```
## Directory Layout
```
app/
├── main.py # FastAPI app, middleware, startup/shutdown
├── core/ # Shared infrastructure
│ ├── config.py # Settings (pydantic-settings)
│ ├── deps.py # Dependency injection
│ ├── security.py # JWT, password hashing
│ └── exceptions.py # Custom exceptions
├── db/
│ ├── base.py # SQLAlchemy Base
│ ├── models.py # DB model imports
│ └── session.py # Engine, SessionLocal
└── {domain}/ # Feature modules
├── models.py # SQLAlchemy models
├── schemas.py # Pydantic schemas
├── service.py # Business logic
└── router.py # API endpoints
```
## Router Pattern
```python
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.deps import get_db, get_current_user
from app.users.models import User
from .schemas import ItemCreate, ItemResponse
from .service import ItemService
router = APIRouter(prefix="/items", tags=["items"])
@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(
data: ItemCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> ItemResponse:
service = ItemService(db)
return await service.create(data, user_id=current_user.id)
```
## Service Pattern
```python
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
class ItemService:
def __init__(self, db: AsyncSession):
self.db = db
async def create(self, data: ItemCreate, user_id: int) -> Item:
item = Item(**data.model_dump(), user_id=user_id)
self.db.add(item)
await self.db.commit()
await self.db.refresh(item)
return item
async def get_by_id(self, item_id: int) -> Item | None:
result = await self.db.execute(select(Item).where(Item.id == item_id))
return result.scalar_one_or_none()
```
## Dependency Injection
```python
# app/core/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import async_session_maker
from app.core.security import decode_access_token
security = HTTPBearer()
async def get_db() -> AsyncSession:
async with async_session_maker() as session:
yield session
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db),
) -> User:
payload = decode_access_token(credentials.credentials)
if not payload:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
user = await db.get(User, payload["sub"])
if not user:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
return user
```
## Pydantic Schemas
```python
from pydantic import BaseModel, Field
from datetime import datetime
# Input
class ItemCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
amount: Decimal = Field(..., gt=0)
# Output
class ItemResponse(BaseModel):
id: int
name: str
amount: Decimal
created_at: datetime
model_config = {"from_attributes": True}
```
## SQLAlchemy Models
```python
from sqlalchemy import Column, Integer, String, ForeignKey, Numeric, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
amount = Column(Numeric(18, 2), nullable=False)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
user = relationship("User", back_populates="items")
```
## Template Method Pattern (Inheritance)
```python
from abc import ABC, abstractmethod
class BaseTransactionService(ABC):
def __init__(self, db: AsyncSession, bank_client: BankClient):
self.db = db
self.bank_client = bank_client
async def process(self, data, user_id: int):
# Common flow
user = await self._get_user(user_id)
self._validate_amount(data.amount)
async with self.db.begin():
await self._execute_bank_operation(data, user)
await self._update_balance(user, data.amount)
transaction = await self._create_transaction(data, user)
return transaction
@abstractmethod
async def _execute_bank_operation(self, data, user):
"""Subclass implements specific bank operation."""
pass
@abstractmethod
async def _update_balance(self, user, amount):
"""Subclass implements balance change direction."""
pass
class DepositService(BaseTransactionService):
async def _execute_bank_operation(self, data, user):
return await self.bank_client.deposit(data)
async def _update_balance(self, user, amount):
user.balance += amount
class WithdrawalService(BaseTransactionService):
async def _execute_bank_operation(self, data, user):
return await self.bank_client.withdraw(data)
async def _update_balance(self, user, amount):
user.balance -= amount
```
## Exception Handling
```python
from fastapi import Request
from fastapi.responses import JSONResponse
class AppException(Exception):
def __init__(self, message: str, status_code: int = 400):
self.message = message
self.status_code = status_code
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.message}
)
```
## This Project
- Auth: `app/auth/` - JWT tokens, login/register
- Users: `app/users/` - User CRUD, account management
- Deposits: `app/deposits/` - Deposit processing
- Withdrawals: `app/withdrawals/` - Withdrawal processing
- Transactions: `app/transactions/` - Transaction history
- Webhooks: `app/webhooks/` - Bank callback handling