1. 创建项目结构
首先，设计一个合理的项目结构。这里以三个Apps为例：users、items 和 orders。
2. 项目结构
my_fastapi_project/
├── main.py
├── app/
│ ├── __init__.py
│ ├── users/
│ │ ├── __init__.py
│ │ ├── models.py
│ │ ├── schemas.py
│ │ ├── crud.py
│ │ └── routes.py
│ ├── items/
│ │ ├── __init__.py
│ │ ├── models.py
│ │ ├── schemas.py
│ │ ├── crud.py
│ │ └── routes.py
│ ├── orders/
│ │ ├── __init__.py
│ │ ├── models.py
│ │ ├── schemas.py
│ │ ├── crud.py
│ │ └── routes.py
├── database.py
└── config.py
3. 配置数据库连接 (database.py)
from databases import Database
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, MetaDataDATABASE_URL = "sqlite:///./test.db" # 使用SQLite作为示例数据库# 初始化数据库连接
database = Database(DATABASE_URL)
metadata = MetaData()
engine = create_engine(DATABASE_URL)Base = declarative_base()async def init_db():async with engine.begin() as conn:await conn.run_sync(Base.metadata.create_all)# 在实际部署中应更加谨慎地处理这一点
import asyncio
asyncio.run(init_db())
4. 定义配置 (config.py)
class Settings:openai_api_key: str = "your-openai-api-key"# 其他配置项...settings = Settings()
5. 每个App的具体实现
app/users/models.py - 用户模型定义
from sqlalchemy import Column, Integer, String
from ..database import Baseclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)username = Column(String, unique=True, index=True)email = Column(String, unique=True, index=True)
6. app/users/schemas.py - 用户Pydantic模式定义
from pydantic import BaseModelclass UserCreate(BaseModel):username: stremail: strclass UserSchema(UserCreate):id: intclass Config:orm_mode = True
7. app/users/crud.py - 用户CRUD操作
from sqlalchemy.orm import Session
from .models import User as UserModel
from .schemas import UserCreateasync def get_user(db: Session, user_id: int):return db.query(UserModel).filter(UserModel.id == user_id).first()async def create_user(db: Session, user: UserCreate):db_user = UserModel(**user.dict())db.add(db_user)await db.commit()await db.refresh(db_user)return db_user
8. app/users/routes.py - 用户路由定义
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ..database import get_db_session
from .crud import get_user, create_user
from .schemas import UserCreate, UserSchemarouter = APIRouter()@router.get("/users/{user_id}", response_model=UserSchema)
async def read_user(user_id: int, db: Session = Depends(get_db_session)):db_user = await get_user(db, user_id=user_id)if db_user is None:raise HTTPException(status_code=404, detail="User not found")return db_user@router.post("/users/", response_model=UserSchema)
async def create_new_user(user: UserCreate, db: Session = Depends(get_db_session)):return await create_user(db=db, user=user)
9. 将所有路由注册到主应用 (main.py)
from fastapi import FastAPI
from app.users.routes import router as users_router
from app.items.routes import router as items_router
from app.orders.routes import router as orders_router
from database import databaseapp = FastAPI()# 包含来自不同Apps的路由
app.include_router(users_router, prefix="/users", tags=["users"])
app.include_router(items_router, prefix="/items", tags=["items"])
app.include_router(orders_router, prefix="/orders", tags=["orders"])@app.on_event("startup")
async def startup():await database.connect()@app.on_event("shutdown")
async def shutdown():await database.disconnect()
9. 启动应用程序
使用uvicorn启动应用程序：
uvicorn main:app --reload