FastAPI + GraphQL 项目架构

FastAPI + GraphQL 项目架构

本项目展示了一个结合 FastAPI 和 GraphQL 的可扩展架构，用于构建现代化 API。

架构优点

1. 高性能和异步支持

   • FastAPI 基于 Starlette 和 Pydantic，提供极高的性能
   • 原生支持异步编程，能够处理高并发请求
   • 自动生成 OpenAPI (Swagger) 文档

2. 灵活的数据查询

   • GraphQL 允许客户端精确指定所需数据
   • 减少网络传输和过度获取问题
   • 单个请求即可获取多个资源

3. 强类型系统

   • Pydantic 提供运行时数据验证
   • GraphQL Schema 确保类型安全
   • Python 类型提示增强代码可维护性

4. 模块化设计

   • 清晰的目录结构便于扩展
   • 关注点分离原则的良好实践
   • 便于团队协作和代码维护

5. 安全性

   • 内置的认证和授权机制
   • 环境变量配置管理
   • 密码加密和 JWT 令牌支持

架构缺点

1. 学习曲线

   • 需要同时掌握 FastAPI 和 GraphQL
   • 理解异步编程概念
   • 需要熟悉多个相关技术栈

2. 复杂性

   • 相比 REST API 架构较为复杂
   • 初期开发速度可能较慢
   • 需要更多的前期设计和规划

3. 性能考虑

   • GraphQL 查询可能导致 N+1 问题
   • 复杂查询可能影响性能
   • 需要合理设计数据加载策略

4. 维护成本

   • Schema 变更需要谨慎处理
   • 需要同时维护 GraphQL 和数据库模型
   • 测试覆盖度要求较高

5. 部署考虑

   • 需要合适的缓存策略
   • 监控和日志更复杂
   • 需要考虑查询复杂度限制

项目结构

my_fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── config.py
│   │   ├── security.py
│   ├── db/
│   │   ├── models.py
│   │   ├── session.py
│   ├── graphql/
│   │   ├── schemas/
│   │   ├── resolvers/
│   ├── services/
│   ├── utils/
├── requirements.txt
├── .env

功能特点

• FastAPI 高性能 Web 框架
• 使用 Strawberry 集成 GraphQL
• SQLAlchemy 数据库 ORM
• Pydantic 数据验证
• JWT 身份认证
• 模块化和可扩展架构

安装设置

1. 创建虚拟环境：

python -m venv venv
source venv/bin/activate  # Windows系统使用: venv\Scripts\activate

2. 安装依赖：

pip install -r requirements.txt

3. 配置环境变量：
   创建 .env 文件并添加以下内容：

DATABASE_URL=sqlite:///./sql_app.db
SECRET_KEY=your-secret-key

4. 运行应用：

uvicorn app.main:app --reload

API 文档

应用运行后，可访问：

• GraphQL 交互界面：http://localhost:8000/graphql
• FastAPI Swagger 文档：http://localhost:8000/docs

项目组件

核心模块

• 配置管理
• 安全认证

数据库

• SQLAlchemy 模型
• 数据库会话管理

GraphQL

• Schema 定义
• 数据查询解析器

服务层

• 业务逻辑
• 数据处理

如何贡献

1. Fork 本仓库
2. 创建特性分支
3. 提交 Pull Request

代码详细说明

1. 核心配置 (core/config.py)

class Settings(BaseSettings):
    APP_NAME: str = "FastAPI + GraphQL Project"
    DATABASE_URL: str = "sqlite:///./sql_app.db"
    SECRET_KEY: str = "your-secret-key-here"

• 使用 Pydantic 的 BaseSettings 管理配置
• 支持从环境变量加载配置
• 提供类型安全的配置访问

2. 数据库模型 (db/models.py)

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    username = Column(String, unique=True, index=True)

• 使用 SQLAlchemy ORM 定义数据模型
• 包含字段验证和索引
• 支持关系映射和级联操作

3. 数据库会话 (db/session.py)

engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

• 创建数据库连接池
• 管理数据库会话生命周期
• 确保资源正确释放

4. GraphQL Schema (graphql/schemas/user_schema.py)

@strawberry.type
class User:
    id: int
    email: str
    username: str

@strawberry.type
class Query:
    @strawberry.field
    def users(self) -> List[User]:
        db = next(get_db())
        return db.query(UserModel).all()

• 定义 GraphQL 类型和查询
• 整合数据库操作
• 支持字段解析和数据验证

5. 安全工具 (core/security.py)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

• JWT 令牌生成和验证
• 密码加密和验证
• 安全会话管理

6. 主应用程序 (main.py)

app = FastAPI(title=settings.APP_NAME)
app.add_middleware(CORSMiddleware, allow_origins=["*"])
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

• FastAPI 应用程序配置
• CORS 中间件设置
• GraphQL 路由集成

代码最佳实践

1. 依赖注入

   • 使用 FastAPI 的依赖注入系统
   • 简化测试和模拟
   • 提高代码重用性

2. 异步操作

@app.get("/async-example")
async def async_example():
    result = await some_async_operation()
    return {"data": result}

• 利用异步特性提高性能
• 处理并发请求
• 避免阻塞操作

3. 错误处理

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    item = await get_item(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

• 统一的错误处理机制
• 清晰的错误信息
• 适当的状态码

4. GraphQL 查询优化

@strawberry.field
def items(self, info: Info, limit: Optional[int] = 10) -> List[Item]:
    query = select(Item).limit(limit)
    return db.scalars(query).all()

• 分页和限制
• 字段选择
• N+1 问题处理

5. 数据验证

from pydantic import BaseModel, EmailStr

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

    class Config:
        orm_mode = True

• 使用 Pydantic 模型
• 输入验证
• 类型转换