当你的应用逐渐发展壮大，业务逻辑变得愈发复杂，模块间需要更细粒度的解耦与协作时，你可能会发现当前这种结构会略显不足。此时，我们便会更推荐采用一种更具扩展性的三层架构——路由层（Router）、服务层（Service）和数据仓库层（Repository）：

路由层 (Router)：负责接收HTTP请求，进行初步的参数验证，并协调调用服务层的方法。它专注于API接口的定义和请求-响应的处理，保持轻量。
服务层 (Service)：承载核心业务逻辑。它将调用数据仓库层来执行数据库操作，并在此基础上处理复杂的业务规则、数据转换等，保持与HTTP层和数据库层的解耦。
数据仓库层 (Repository)：专注于与数据库的交互，封装所有的CRUD（创建、读取、更新、删除）操作，对外提供统一的数据访问接口，使得业务逻辑层无需关心具体的ORM细节，例如本示例中的crud模块就可以演变为repository层的一部分。

.
├── app/
│   ├── __init__.py             # 初始化模块
│   ├── main.py                 # 应用入口文件
│   ├── core/                   # 核心配置和工具模块
│   │   ├── __init__.py
│   │   ├── config.py           # 配置文件，如数据库连接、JWT 密钥等
│   │   ├── security.py         # 安全相关工具（如密码加密、JWT 生成与验证）
│   │   └── dependencies.py     # 全局依赖项
│   ├── models/                 # 数据库模型
│   │   ├── __init__.py
│   │   ├── base.py             # 基础模型和数据库连接
│   │   └── user.py             # 用户模型
│   ├── schemas/                # 数据验证和序列化
│   │   ├── __init__.py
│   │   ├── user.py             # 用户相关 Pydantic 模型
│   ├── crud/                   # 数据库操作（CRUD）
│   │   ├── __init__.py
│   │   └── user.py             # 用户相关 CRUD 操作
│   ├── api/                    # API 路由
│   │   ├── __init__.py
│   │   ├── deps.py             # API 路由依赖项
│   │   ├── user.py             # 用户相关路由
│   │   └── auth.py             # 身份验证路由
│   ├── tests/                  # 测试用例
│   │   ├── __init__.py
│   │   ├── test_auth.py        # 身份验证相关测试
│   │   └── test_user.py        # 用户相关测试
├── .env                        # 环境变量
├── requirements.txt            # 项目依赖
└── alembic/                    # 数据库迁移（可选）
    ├── env.py
    └── versions/

main.py

from fastapi import FastAPI
from app.api import auth, user

app = FastAPI()

# 注册路由
app.include_router(auth.router, prefix="/auth", tags=["Authentication"])
app.include_router(user.router, prefix="/users", tags=["Users"])

core/config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    JWT_SECRET: str = "your_jwt_secret"
    JWT_ALGORITHM: str = "HS256"
    DATABASE_URL: str = "sqlite+aiosqlite:///./test.db"

    class Config:
        env_file = ".env"

settings = Settings()

core/security.py

from passlib.context import CryptContext
from jose import jwt
from datetime import datetime, timedelta
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: timedelta = timedelta(minutes=30)):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)

models/base.py

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from app.core.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

models/user.py

from sqlalchemy import String, Integer, Boolean
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String, nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

schemas/user.py

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    id: int
    email: EmailStr
    is_active: bool

    model_config = ConfigDict(from_attributes=True)

crud/user.py

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.models.user import User
from app.schemas.user import UserCreate
from app.core.security import hash_password

async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).filter(User.email == email))
    return result.scalars().first()

async def create_user(db: AsyncSession, user: UserCreate) -> User:
    db_user = User(email=user.email, hashed_password=hash_password(user.password))
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

api/auth.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.schemas.user import UserCreate
from app.core.security import verify_password, create_access_token
from app.crud.user import get_user_by_email
from app.core.dependencies import get_db

router = APIRouter()

@router.post("/login")
async def login(email: str, password: str, db: AsyncSession = Depends(get_db)):
    user = await get_user_by_email(db, email)
    if not user or not verify_password(password, user.hashed_password):
        raise HTTPException(status_code=400, detail="Invalid credentials")
    token = create_access_token(data={"sub": user.email})
    return {"access_token": token, "token_type": "bearer"}

core/dependencies.py

from app.models.base import async_session

async def get_db():
    async with async_session() as session:
        yield session