
fastapi_project/├── main.py # 程序入口,注册路由、启动app├── core/ # 核心配置│ ├── config.py # 全局配置(数据库地址、密钥、跨域)│ ├── security.py # JWT、密码加密、权限校验│ └── dependencies.py # 全局依赖(登录鉴权、分页、日志)├── db/ # 数据库相关│ ├── base.py # 数据库基类、会话创建│ ├── session.py# 数据库连接、get_db依赖│ └── models/ # SQLAlchemy ORM模型│ ├── user.py│ └── goods.py├── crud/ # 数据库操作封装(增删改查)│ ├── base.py # 通用CRUD父类│ ├── user_crud.py│ └── goods_crud.py├── schemas/# Pydantic 序列化/校验模型│ ├── user.py # 用户入参、出参模型│ └── goods.py├── routers/# 接口路由(按模块拆分)│ ├── __init__.py # 统一汇总所有路由│ ├── user.py # 用户模块接口│ └── goods.py# 商品模块接口├── services/ # 业务逻辑层(复杂逻辑放这里)│ ├── user_service.py│ └── order_service.py├── utils/# 通用工具│ ├── logger.py│ ├── file.py│ └── time.py├── static/ # 静态文件├── tests/# 单元测试└── requirements.txt
from fastapi import FastAPIfrom core.config import settingsfrom core.dependencies import cors_middlewarefrom routers import api_routerapp = FastAPI(title=settings.PROJECT_NAME)# 跨域app.add_middleware(cors_middleware)# 挂载所有路由app.include_router(api_router, prefix="/api/v1")if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)
from fastapi import APIRouter, Dependsfrom schemas.user import UserCreate, UserRespfrom services.user_service import create_userfrom db.session import get_dbfrom sqlalchemy.orm import Sessionrouter = APIRouter(prefix="/user", tags=["用户模块"])@router.post("/add", response_model=UserResp)def add_user(user_in: UserCreate, db: Session = Depends(get_db)):# 只调用service,不写业务return create_user(db=db, user_in=user_in)
routers/init.py 聚合路由python
运行
from fastapi import APIRouterfrom .user import router as user_routerfrom .goods import router as goods_routerapi_router = APIRouter()api_router.include_router(user_router)api_router.include_router(goods_router)
from sqlalchemy.orm import Sessionfrom schemas.user import UserCreatefrom crud.user_crud import user_crudfrom core.security import hash_passworddef create_user(db: Session, user_in: UserCreate):# 业务逻辑:密码加密、重复校验、联动其他表user_in.password = hash_password(user_in.password)return user_crud.create(db, obj_in=user_in)
from crud.base import CRUDBasefrom db.models.user import Userfrom schemas.user import UserCreate, UserUpdateclass CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):def get_by_phone(self, db: Session, phone: str):return db.query(User).filter(User.phone == phone).first()user_crud = CRUDUser(User)
get_dbbase:公共模型基类(id、创建时间、更新时间)routers/user、services/user、crud/user、schemas/usermain → routers → services → crud → db.modelsclients/ 文件夹存放第三方 SDK(支付、OSS、短信)定时任务:单独 tasks/,使用 APScheduler,不和接口代码耦合