当前位置：首页 > article >正文

FastAPI中创建一个多App架构

article 2025/3/5 6:06:57

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, MetaData

DATABASE_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 Base

class 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 BaseModel

class UserCreate(BaseModel):
    username: str
    email: str

class UserSchema(UserCreate):
    id: int

    class Config:
        orm_mode = True

7. app/users/crud.py - 用户CRUD操作

from sqlalchemy.orm import Session
from .models import User as UserModel
from .schemas import UserCreate

async 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, UserSchema

router = 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 database

app = 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()