智创空间(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/ # 外部参考文档(百炼平台)
main.py - 应用入口功能: 应用程序的主入口,负责创建和配置 FastAPI 应用实例。
# 主要组件说明
| 组件 | 说明 |
|---|---|
lifespan() |
异步上下文管理器,管理应用生命周期,启动时验证数据库连接 |
app |
FastAPI 应用实例,配置了标题、描述、版本信息 |
CORSMiddleware |
跨域资源共享中间件,允许所有来源访问 |
health_check() |
健康检查端点,返回服务状态 |
关键配置:
APP_HOST、APP_PORT、DEBUG0.0.0.0:8000app/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: 数据库会话对象
"""
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}')>"
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] | 响应数据 |
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): 模型不存在时抛出
"""
app/routers/model_router.py - API路由层功能: 提供模型数据的 RESTful API 端点。
路由前缀: /api/models
标签: 模型广场
/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": [...]
}
}
/api/models/{model_id}功能: 根据ID获取单个模型详情
| 参数 | 类型 | 说明 |
|---|---|---|
model_id |
int | 模型ID(路径参数) |
响应: ApiResponse[ModelResponse]
{
"code": 200,
"message": "success",
"data": {
"id": 1,
"title": "qwen-turbo",
"name": "通义千问",
...
}
}
app/middleware/error_handler.py - 异常处理功能: 处理各类异常并返回统一格式的错误响应。
register_exception_handlers(app: FastAPI)功能: 注册全局异常处理器到 FastAPI 应用
| 异常类型 | HTTP状态码 | 说明 |
|---|---|---|
HTTPException |
原状态码 | HTTP异常处理 |
RequestValidationError |
400 | 参数验证错误 |
SQLAlchemyError |
500 | 数据库错误 |
Exception |
500 | 全局未捕获异常 |
统一响应格式:
{
"code": <状态码>,
"message": "<错误信息>",
"data": null
}
scripts/init_data.py - 数据初始化脚本功能: 从 JSON 文件导入模型数据到数据库。
init_data()def init_data():
"""
初始化模型数据
数据源: ../模型列表.json
处理逻辑:
1. 读取JSON文件
2. 遍历每条数据
3. 根据title判断是否存在
- 存在: 更新记录
- 不存在: 插入新记录
4. 提交事务
5. 输出统计信息
"""
使用方式:
python scripts/init_data.py
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功能: 数据库连接测试脚本
测试内容:
| 包名 | 版本 | 说明 |
|---|---|---|
| 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 │
│ (数据验证) │ │ (会话管理) │ │ (数据库) │
└─────────────┘ └─────────────┘ └─────────────┘
pip install -r requirements.txt
创建 .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
# 执行迁移脚本
psql -U postgres -d model_square -f migrations/001_create_models_table.sql
# 导入初始数据
python scripts/init_data.py
python main.py
# 或
uvicorn main:app --reload
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /health |
健康检查 |
| GET | /api/models |
获取模型列表(支持分页和搜索) |
| GET | /api/models/{model_id} |
获取单个模型详情 |
ApiResponse 格式PaginatedResponse[T] 和 ApiResponse[T] 支持类型复用文档生成时间: 2025年12月26日