project.md 16 KB

AIGC-Space 后端项目详细文档

📋 项目概述

智创空间(AIGC-Space) 是一个AI模型广场后端服务,基于 FastAPI 框架构建,提供模型信息的展示、搜索和管理功能。项目采用经典的分层架构设计,使用 PostgreSQL 作为数据存储,SQLAlchemy 作为 ORM 框架。


🏗️ 项目结构

backend/
├── main.py                     # 应用入口文件
├── requirements.txt            # Python依赖包列表
├── .env                        # 环境变量配置(需自行创建)
│
├── app/                        # 应用核心代码
│   ├── __init__.py            # 包初始化文件
│   ├── database.py            # 数据库配置与连接管理
│   │
│   ├── models/                # ORM模型层
│   │   ├── __init__.py
│   │   └── model.py           # AI模型ORM定义
│   │
│   ├── schemas/               # 数据传输对象层
│   │   ├── __init__.py
│   │   └── model_schema.py    # 请求/响应数据结构
│   │
│   ├── services/              # 业务逻辑层
│   │   ├── __init__.py
│   │   └── model_service.py   # 模型业务服务
│   │
│   ├── routers/               # API路由层
│   │   ├── __init__.py
│   │   └── model_router.py    # 模型API端点
│   │
│   └── middleware/            # 中间件层
│       ├── __init__.py
│       └── error_handler.py   # 全局异常处理
│
├── migrations/                 # 数据库迁移脚本
│   └── 001_create_models_table.sql
│
├── scripts/                    # 工具脚本
│   ├── __init__.py
│   └── init_data.py           # 数据初始化脚本
│
├── tests/                      # 测试代码
│   ├── __init__.py
│   ├── test_api_integration.py    # API集成测试
│   ├── test_db_connection.py      # 数据库连接测试
│   ├── test_model_structure.py    # 模型结构测试
│   ├── test_property_error.py     # 属性错误测试
│   └── test_property_session.py   # 会话属性测试
│
└── docs/                       # 文档目录
    ├── generate_prompt/        # 生成提示词文档
    ├── Inside/                 # 内部API文档
    └── Outside/                # 外部参考文档(百炼平台)

📁 核心文件详解

1. main.py - 应用入口

功能: 应用程序的主入口,负责创建和配置 FastAPI 应用实例。

# 主要组件说明
组件 说明
lifespan() 异步上下文管理器,管理应用生命周期,启动时验证数据库连接
app FastAPI 应用实例,配置了标题、描述、版本信息
CORSMiddleware 跨域资源共享中间件,允许所有来源访问
health_check() 健康检查端点,返回服务状态

关键配置:

  • 支持通过环境变量配置 APP_HOSTAPP_PORTDEBUG
  • 默认监听 0.0.0.0:8000
  • 开发模式下启用热重载

2. app/database.py - 数据库配置模块

功能: 提供数据库连接、会话管理和依赖注入功能。

配置变量

变量名 默认值 说明
DB_HOST localhost 数据库主机地址
DB_PORT 5432 数据库端口
DB_USER postgres 数据库用户名
DB_PASSWORD (空) 数据库密码
DB_NAME model_square 数据库名称
DB_POOL_SIZE 10 连接池大小
DB_MAX_OVERFLOW 20 最大溢出连接数

核心对象

对象 类型 说明
engine Engine SQLAlchemy 数据库引擎,配置了连接池和预检查
SessionLocal sessionmaker 会话工厂,用于创建数据库会话
Base declarative_base ORM 模型基类

函数

get_db()
def get_db():
    """
    数据库会话依赖注入函数
    
    用于FastAPI的依赖注入系统,自动管理数据库会话的生命周期
    确保请求结束后会话被正确关闭
    
    Yields:
        Session: 数据库会话对象
    """

3. app/models/model.py - ORM模型定义

功能: 定义 AI 模型的数据库表结构。

Model

继承: Base (SQLAlchemy declarative_base)

表名: models

Schema: aigcspace

字段定义
字段名 类型 约束 说明
id Integer 主键, 自增 主键ID
title String(255) 非空, 唯一, 索引 模型标识
img Text 非空 图标URL
name Text 非空 显示名称
tag1 String(100) 可空 主标签
description Text 可空 模型描述
keyword Text 可空, 索引 关键词(用于搜索)
time String(50) 可空 发布时间
tag2 String(100) 可空 次要标签
created_at DateTime 非空, 自动设置 创建时间
updated_at DateTime 非空, 自动更新 更新时间
方法
def __repr__(self):
    """返回模型的字符串表示"""
    return f"<Model(id={self.id}, title='{self.title}', name='{self.name}')>"

4. app/schemas/model_schema.py - 数据传输对象

功能: 定义 API 请求和响应的数据结构(Pydantic 模型)。

ModelResponse

功能: 模型响应数据结构

字段 类型 说明
id int 模型ID
title str 模型标识
img str 图标URL
name str 显示名称
tag1 Optional[str] 主标签
description Optional[str] 模型描述
keyword Optional[str] 关键词
time Optional[str] 发布时间
tag2 Optional[str] 次要标签
created_at datetime 创建时间
updated_at datetime 更新时间

配置: from_attributes=True 支持从 ORM 对象直接转换


PaginatedResponse[T]

功能: 泛型分页响应数据结构

字段 类型 说明
total int 总记录数
page int 当前页码
page_size int 每页大小
items List[T] 数据列表

ApiResponse[T]

功能: 统一 API 响应格式

字段 类型 说明
code int 状态码
message str 响应消息
data Optional[T] 响应数据

5. app/services/model_service.py - 业务服务层

功能: 提供模型数据的业务逻辑处理。

ModelService

构造函数
def __init__(self, db: Session):
    """
    初始化服务
    
    Args:
        db: SQLAlchemy 数据库会话
    """
    self.db = db
方法详解
get_models()
def get_models(self, page: int = 1, page_size: int = 20) -> PaginatedResponse[ModelResponse]:
    """
    获取分页的模型列表
    
    Args:
        page: 页码,默认1
        page_size: 每页大小,默认20
    
    Returns:
        PaginatedResponse[ModelResponse]: 分页响应数据
    
    实现逻辑:
        1. 查询总记录数
        2. 计算偏移量 offset = (page - 1) * page_size
        3. 执行分页查询
        4. 将ORM对象转换为响应模型
    """
search_models()
def search_models(
    self, 
    keyword: Optional[str] = None, 
    page: int = 1, 
    page_size: int = 20
) -> PaginatedResponse[ModelResponse]:
    """
    根据关键词搜索模型
    
    Args:
        keyword: 搜索关键词
        page: 页码,默认1
        page_size: 每页大小,默认20
    
    Returns:
        PaginatedResponse[ModelResponse]: 分页响应数据
    
    搜索范围:
        - name: 模型名称
        - description: 模型描述
        - keyword: 关键词字段
    
    搜索方式: 模糊匹配 (ILIKE %keyword%)
    """
get_model_by_id()
def get_model_by_id(self, model_id: int) -> ModelResponse:
    """
    根据ID获取单个模型
    
    Args:
        model_id: 模型ID
    
    Returns:
        ModelResponse: 模型响应数据
    
    Raises:
        HTTPException(404): 模型不存在时抛出
    """

6. app/routers/model_router.py - API路由层

功能: 提供模型数据的 RESTful API 端点。

路由前缀: /api/models

标签: 模型广场

API 端点

GET /api/models

功能: 获取模型列表(支持分页和搜索)

参数 类型 默认值 约束 说明
page int 1 >= 1 页码
page_size int 20 1-100 每页大小
keyword str None - 搜索关键词

响应: ApiResponse[PaginatedResponse[ModelResponse]]

{
    "code": 200,
    "message": "success",
    "data": {
        "total": 100,
        "page": 1,
        "page_size": 20,
        "items": [...]
    }
}

GET /api/models/{model_id}

功能: 根据ID获取单个模型详情

参数 类型 说明
model_id int 模型ID(路径参数)

响应: ApiResponse[ModelResponse]

{
    "code": 200,
    "message": "success",
    "data": {
        "id": 1,
        "title": "qwen-turbo",
        "name": "通义千问",
        ...
    }
}

7. app/middleware/error_handler.py - 异常处理

功能: 处理各类异常并返回统一格式的错误响应。

register_exception_handlers(app: FastAPI)

功能: 注册全局异常处理器到 FastAPI 应用

异常处理器
异常类型 HTTP状态码 说明
HTTPException 原状态码 HTTP异常处理
RequestValidationError 400 参数验证错误
SQLAlchemyError 500 数据库错误
Exception 500 全局未捕获异常

统一响应格式:

{
    "code": <状态码>,
    "message": "<错误信息>",
    "data": null
}

8. scripts/init_data.py - 数据初始化脚本

功能: 从 JSON 文件导入模型数据到数据库。

init_data()

def init_data():
    """
    初始化模型数据
    
    数据源: ../模型列表.json
    
    处理逻辑:
        1. 读取JSON文件
        2. 遍历每条数据
        3. 根据title判断是否存在
           - 存在: 更新记录
           - 不存在: 插入新记录
        4. 提交事务
        5. 输出统计信息
    """

使用方式:

python scripts/init_data.py

9. migrations/001_create_models_table.sql - 数据库迁移

功能: 创建 AI 模型信息表的 SQL 脚本。

包含内容:

  • 创建 models
  • 添加表和字段注释
  • 创建 keyword 索引
  • 回滚脚本(注释状态)

🧪 测试模块

tests/test_api_integration.py

功能: API 集成测试

TestModelAPI

测试方法 说明
test_get_models_success() 测试获取模型列表成功
test_get_models_with_pagination() 测试分页参数
test_get_models_with_keyword() 测试关键词搜索
test_get_model_not_found() 测试获取不存在的模型
test_invalid_page_parameter() 测试无效的page参数
test_invalid_page_size_parameter() 测试无效的page_size参数
test_health_check() 测试健康检查端点

tests/test_model_structure.py

功能: 模型结构验证测试

TestModelStructure

测试方法 说明
test_table_name() 验证表名为models
test_all_fields_exist() 验证所有必需字段存在
test_field_types() 验证字段类型正确
test_primary_key() 验证id为主键
test_title_unique_index() 验证title字段有唯一索引
test_keyword_index() 验证keyword字段有索引
test_nullable_constraints() 验证字段的nullable约束

tests/test_db_connection.py

功能: 数据库连接测试脚本

测试内容:

  • 数据库连接是否成功
  • 连接池配置是否正确
  • 会话创建和关闭是否正常

📦 依赖说明

Web框架

包名 版本 说明
fastapi 0.104.1 现代高性能Web框架
uvicorn[standard] 0.24.0 ASGI服务器

数据库

包名 版本 说明
sqlalchemy 2.0.23 ORM框架
psycopg2-binary 2.9.9 PostgreSQL驱动

数据验证

包名 版本 说明
pydantic 2.5.0 数据验证库
pydantic-settings 2.1.0 配置管理

工具

包名 版本 说明
python-dotenv 1.0.0 环境变量管理

测试

包名 版本 说明
pytest 7.4.3 测试框架
pytest-asyncio 0.21.1 异步测试支持
hypothesis 6.92.1 属性测试
httpx 0.25.2 HTTP客户端

🔄 数据流程图

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Client    │────▶│   Router    │────▶│   Service   │────▶│   Model     │
│  (请求)     │     │  (路由层)    │     │  (业务层)    │     │  (数据层)   │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
                           │                   │                   │
                           ▼                   ▼                   ▼
                    ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
                    │   Schema    │     │  Database   │     │ PostgreSQL  │
                    │ (数据验证)   │     │  (会话管理)  │     │  (数据库)   │
                    └─────────────┘     └─────────────┘     └─────────────┘

🚀 快速开始

1. 安装依赖

pip install -r requirements.txt

2. 配置环境变量

创建 .env 文件:

DB_HOST=localhost
DB_PORT=5432
DB_USER=postgres
DB_PASSWORD=your_password
DB_NAME=model_square
APP_HOST=0.0.0.0
APP_PORT=8000
DEBUG=True

3. 初始化数据库

# 执行迁移脚本
psql -U postgres -d model_square -f migrations/001_create_models_table.sql

# 导入初始数据
python scripts/init_data.py

4. 启动服务

python main.py
# 或
uvicorn main:app --reload

5. 访问API文档


📝 API 接口汇总

方法 路径 说明
GET /health 健康检查
GET /api/models 获取模型列表(支持分页和搜索)
GET /api/models/{model_id} 获取单个模型详情

🏛️ 架构设计原则

  1. 分层架构: Router → Service → Model,职责清晰
  2. 依赖注入: 使用 FastAPI 的 Depends 实现数据库会话注入
  3. 统一响应: 所有 API 返回统一的 ApiResponse 格式
  4. 异常处理: 全局异常处理器确保错误响应格式一致
  5. 类型安全: 使用 Pydantic 进行数据验证和序列化
  6. 泛型支持: PaginatedResponse[T]ApiResponse[T] 支持类型复用

文档生成时间: 2025年12月26日