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

FastAPI中创建一个多App架构

news 2025/7/28 12:49:02

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