技术栈与设计理念
技术栈总览
| 技术 | 版本 | 用途 |
|---|---|---|
| FastAPI | 0.121.1 | 异步 Web 框架 |
| SQLAlchemy | 2.0.23 | 异步 ORM |
| Pydantic | 2.5.2 | 数据验证与序列化 |
| asyncpg | 0.29.0 | PostgreSQL 异步驱动 |
| aiomysql | 0.2.0 | MySQL 异步驱动 |
| Redis | 5.0.1 | 缓存/会话/任务队列 |
| APScheduler | 4.0.0a6 | 定时任务调度 |
| Alembic | 1.13.0 | 数据库迁移 |
| OpenAI SDK | 2.9.0 | AI 模型调用 |
| Qdrant | 1.12.1 | 向量数据库 |
| Minio | 7.2.16 | 对象存储 |
设计理念
全异步架构
整个后端采用异步编程模型，从数据库驱动到 HTTP 客户端全链路异步：
FastAPI (ASGI) → AsyncSession (SQLAlchemy) → asyncpg/aiomysql → Database
→ httpx (异步HTTP)
→ aiosmtplib (异步邮件)
→ redis.asyncio (异步Redis)约定优于配置
- BaseModel 提供统一的主键、软删除、审计字段
- BaseService 提供通用 CRUD、Excel 导入导出、数据权限
- ResourceRegistry 自动注册资源类型和字段元数据
- Service 子类只需定义
model属性即可获得完整功能
逻辑外键
不使用数据库外键约束，通过代码逻辑保证数据一致性。优点：
- 避免级联删除导致的意外数据丢失
- 灵活的数据清理策略
- 更好的数据库迁移兼容性
- 支持跨数据库关联
多数据库支持
通过配置 DB_TYPE 切换数据库类型：
| 数据库 | 驱动 | 连接 URL |
|---|---|---|
| PostgreSQL | asyncpg | postgresql+asyncpg:// |
| MySQL | aiomysql | mysql+aiomysql:// |
| SQL Server | aioodbc | mssql+aioodbc:// |
项目结构
backend-fastapi/
├── app/ # 核心应用模块
│ ├── config.py # 配置管理 (Pydantic Settings)
│ ├── database.py # 数据库连接与会话管理
│ ├── base_model.py # BaseModel 基类
│ ├── base_schema.py # 通用 Schema
│ ├── base_service.py # BaseService 基类
│ ├── cache_service.py # 带缓存的 Service 基类
│ ├── resource_registry.py # 资源注册表
│ ├── field_metadata_generator.py # 字段元数据生成器
│ ├── field_permission_cache.py # 字段权限缓存
│ ├── config_manager.py # 配置管理器
│ ├── db_compat.py # 数据库兼容层
│ ├── timezone.py # 时区处理
│ ├── email.py # 邮件服务
│ ├── sms.py # 短信服务
│ ├── dingtalk.py # 钉钉集成
│ ├── feishu.py # 飞书集成
│ ├── wechat.py # 微信集成
│ └── wecom.py # 企业微信集成
│
├── core/ # 业务模块
│ ├── router.py # 统一路由入口
│ ├── auth/ # 认证授权
│ ├── user/ # 用户管理
│ ├── dept/ # 部门管理
│ ├── role/ # 角色管理
│ ├── post/ # 岗位管理
│ ├── menu/ # 菜单管理
│ ├── permission/ # 权限管理
│ ├── dict/ # 数据字典
│ ├── application/ # 应用管理
│ ├── form_manager/ # 表单元数据管理
│ ├── form_data_manager/ # 表单数据动态操作
│ ├── workflow/ # 工作流引擎
│ ├── ai_platform/ # AI 平台
│ ├── file_manager/ # 文件管理
│ ├── page_manager/ # 页面管理
│ ├── data_source/ # 数据源管理
│ ├── notification/ # 消息通知
│ ├── chat/ # 即时通讯
│ ├── ui_config/ # UI 配置
│ └── ...
│
├── alembic/ # 数据库迁移
│ ├── env.py
│ └── versions/ # 迁移脚本
│
├── env/ # 环境配置
│ ├── example.env # 示例配置
│ ├── dev.env # 开发环境
│ ├── uat.env # 测试环境
│ └── prod.env # 生产环境
│
├── main.py # 应用入口
├── requirements.txt # Python 依赖
└── alembic.ini # Alembic 配置配置管理
基于 Pydantic Settings 的配置系统，支持环境变量和 .env 文件：
python
class Settings(BaseSettings):
ENV: Literal["dev", "uat", "prod"] = "dev"
DEBUG: bool = False
# 数据库
DB_TYPE: Literal["postgresql", "mysql", "sqlserver"] = "postgresql"
DB_HOST: str = "localhost"
DB_PORT: int = 5432
DATABASE_URL: Optional[str] = None # 可手动配置，否则自动拼接
# Redis
REDIS_HOST: str = "localhost"
REDIS_URL: Optional[str] = None # 可手动配置，否则自动拼接
# JWT
JWT_SECRET_KEY: str = "..."
ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440 # 24小时
# 文件存储
FILE_STORAGE_TYPE: str = "minio" # local/oss/minio/azure
model_config = SettingsConfigDict(
env_file=f"env/{os.getenv('ENV', 'dev')}.env",
case_sensitive=True,
)自动 URL 拼接：如果未手动配置 DATABASE_URL 或 REDIS_URL，系统根据各配置项自动拼接。
应用入口
main.py 负责：
- 创建 FastAPI 实例
- 配置生命周期（启动时注册已发布表单、初始化 Redis、启动调度器）
- 添加认证中间件
- 挂载路由（core + zq_demo + websocket）
- 配置 CORS
python
app = FastAPI(
title=settings.APP_NAME,
version="1.0.0",
lifespan=lifespan,
)
# 全局认证中间件
app.add_middleware(AuthPermissionMiddleware)
# 路由挂载
app.include_router(core_router, prefix="/api/core", dependencies=[Depends(oauth2_scheme)])