BISHENG 后端开发规范实战指南:基于 FastAPI + SQLAlchemy 的 Python 编码、架构与工程质量标准
【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng
BISHENG 作为面向下一代企业级 AI 应用的开源 LLM DevOps 平台,其 Python 后端(FastAPI + SQLAlchemy)承载了 RAG、Agent、工作流、权限等多达数十个业务模块。为保证代码风格统一、结构清晰且易于维护,仓库内置了一份开发规范 Skill(位于 src/backend/.agents/skills/coding_guidelines/SKILL.md)。本篇指南将以该规范为核心骨架,结合仓库真实源码逐条展开讲解,帮助开发者快速掌握 BISHENG 后端的编码风格、分层架构、数据库迁移、异常处理与日志规范,并能在日常开发与代码评审中直接落地使用。
1. 编码风格:统一代码语言的第一道关卡
规范要求严格遵循PEP 8,并强制使用类型注解:所有函数、方法参数及返回值必须明确指定类型,以支持静态检查和代码智能感知。
1.1 命名规范
| 对象 | 命名风格 | 示例 |
|---|---|---|
| 包名 / 模块名 / 变量名 / 函数名 | snake_case | user_service.py、get_user_by_id |
| 类名 | PascalCase | UserService、BaseErrorCode |
| 常量名 | UPPER_SNAKE_CASE | MAX_RETRY_TIMES、DEFAULT_PAGE_SIZE |
1.2 代码格式化
规范假设项目使用black、ruff或isort进行格式化,开发时须保持与之相符的格式。从仓库的实际代码可以印证这套风格被严格落实:
- 类型注解无处不在,例如 base_repository.py 中
class BaseRepository(ABC, Generic[T, ID])以及async def find_by_id(self, entity_id: ID) -> Optional[T]; - 泛型配合类型变量
T = TypeVar('T', bound=SQLModel)、ID = TypeVar('ID')在 base_repository_impl.py 中定义,体现了对静态检查的深度依赖。
2. 分层架构设计:业务模块的标准化目录结构
BISHENG 后端要求每个业务模块拥有独立的目录,并在目录内部按职责分层。仓库中user、knowledge、permission、channel、tenant等数十个模块均遵循此结构,例如 src/backend/bisheng/user 与 src/backend/bisheng/knowledge。
业务模块目录/ ├── api/ # 与外部交互的接口层 │ ├── endpoints/ # 具体的 API 端点实现(FastAPI 路由) │ ├── dependencies.py # API 层依赖项(如 service 层依赖) │ └── router.py # API 路由定义 ├── domain/ # 核心业务逻辑与领域模型 │ ├── models/ # 数据库模型(SQLModel) │ ├── services/ # 业务服务类,封装核心业务逻辑 │ ├── repositories/ # 数据访问层,封装数据库操作 │ │ ├── implementations/ # 具体 Repository 实现 │ │ └── interfaces/ # Repository 接口定义 │ └── schemas/ # Pydantic 模型,用于数据验证与序列化2.1 Repository 层的接口与实现分离
规范要求implementations下的具体实现应继承BaseRepositoryImpl[ModelClass, IDType], RepositoryInterface,而interfaces下的接口应继承BaseRepository[ModelClass, IDType], ABC。仓库提供了这一体系的公共基类:
- 接口层:BaseRepository 继承
ABC与Generic[T, ID],用@abstractmethod声明了 18 个抽象方法,覆盖完整的 CRUD 能力:save、bulk_save、find_by_id、find_one、find_by_ids、find_all、update、delete、exists、count,且每个方法都同时提供异步版与_sync同步版(如find_by_id_sync),兼顾 async/await 与同步调用场景。 - 实现层:BaseRepositoryImpl 基于 SQLModel 的
Session/AsyncSession实现了上述全部方法。例如save通过session.add(entity)后commit并refresh返回实体;find_one(**filters)使用getattr(self.model_class, field) == value动态构建select查询条件;count则通过func.count()构造聚合查询。
这种"接口定义契约、实现封装细节"的模式,使得业务 Service 层只依赖抽象接口,数据库实现可以灵活替换,也便于单元测试时注入 Mock。
2.2 各层职责边界
- api 层:只负责 HTTP 协议处理、参数校验与响应组装,不包含业务逻辑;
- domain/services 层:承载核心业务规则,调用 Repository 完成数据操作;
- domain/repositories 层:封装所有 SQLModel 查询,屏蔽底层 ORM 细节;
- domain/schemas 层:定义 Pydantic 模型,用于请求体验证与响应序列化。
3. 数据库与 ORM:SQLModel + Alembic 的演进机制
3.1 表结构变更必须走 Alembic 迁移
规范明确:所有数据库表结构的增删改必须通过 Alembic 生成迁移文件,禁止直接手动修改数据库表结构。迁移脚本目录为 src/backend/bisheng/core/database/alembic/versions,仓库内已有从v2_3_0_beta1到v2_6_0的数十个版本化迁移文件(如v2_5_0_f004_rebac.py、v2_6_0_f035_linsight_skill.py),每个文件对应一次经过评审的表结构演进。
官方说明文件 src/backend/bisheng/core/database/alembic/README.md 给出了完整的日常操作命令:
# 创建迁移脚本(手动) alembic revision -m "描述信息" # 自动生成迁移脚本(基于模型与数据库差异,差异较大时建议先手动创建再修改) alembic revision --autogenerate -m "描述信息" # 应用所有未应用的迁移 alembic upgrade head # 升级到指定版本 alembic upgrade <版本号> # 回滚到上一个版本 alembic downgrade -1 # 回滚到指定版本 alembic downgrade <版本号> # 查看当前数据库版本 alembic current # 查看迁移历史 alembic history # 查看未应用的迁移脚本 alembic heads # 仅生成迁移 SQL 文件而不直接执行 alembic upgrade head --sql > upgrade.sql该 README 还说明了 Alembic 的配置文件与目录职责:alembic.ini为全局主配置(含数据库连接串),env.py 负责连接数据库与加载模型元数据,script.py.mako是迁移脚本模板。同时需要特别注意的是其Schema ownership约定:缺失的整表由最新发现的 SQLModel 元数据在升级前自动创建,而已有表的列/索引/约束变更则一律通过 Alembic revision 应用,create_all()不会修改已有表,也不能替代迁移脚本。
3.2 数据访问必须通过 ORM 模型
规范要求所有增删改查必须通过 SQLModel 或 SQLAlchemy ORM 进行,禁止直接使用原生 SQL,会话通过依赖注入或@db_session装饰器管理以保证事务一致性与资源释放。仓库中的 src/backend/bisheng/core/database/manager.py 提供了DatabaseConnectionManager以及get_database_connection()/sync_get_database_connection()等会话获取入口,是依赖注入与连接管理的核心。
3.3 模型定义要点
- 使用 SQLModel 定义模型,确保与数据库表结构一致,并利用其原生 Pydantic 校验能力;
- 模型类放在
domain/models模块中,每个类必须有明确的__tablename__与完整字段定义; - 字段使用 SQLModel 的
Field函数定义,明确类型、默认值、索引等属性; - 需要 SQLAlchemy 特有字段属性时,通过
sa_column等方式定义。
3.4 查询与性能规范
- 尽量使用 SQLModel 查询接口,避免原生 SQL;复杂查询在 Repository 层封装成方法并提供清晰接口;
- 涉及多表查询时,优先使用 JOIN 或子查询,避免在 Python 代码中多次查询后手动拼装数据;
- 分页查询优先使用 SQLModel 分页能力,或在 Repository 层封装通用分页方法以提高复用性(仓库的
find_all/find_by_ids等通用方法即是封装思想的体现); - 批量插入、更新等操作使用批量接口(如
BaseRepositoryImpl.bulk_save中的session.add_all),提升效率与性能。
4. 异常处理与日志:可观测的业务错误体系
4.1 分业务自定义异常,统一继承 BaseErrorCode
规范要求每个业务模块定义自己的异常类,继承自公共基类BaseErrorCode,并按模块在 src/backend/bisheng/common/errcode 目录下拆分文件(如user.py、knowledge.py、approval.py等)。仓库中该目录已包含数十个按业务划分的异常定义文件。
基类定义位于 src/backend/bisheng/common/errcode/base.py:
class BaseErrorCode(Exception): # 错误码前三位代表功能模块,后两位代表模块内的具体错误,例如 10001 Code: int Msg: str def __init__(self, exception: Exception = None, msg: str = None, code: int = None, **kwargs): self.exception = exception self.message = msg or self.Msg self.code = code or self.Code self.kwargs = kwargs super().__init__(exception)BaseErrorCode不仅是一个带Code/Msg属性的异常基类,还内置了丰富的响应输出工具方法:
return_resp()/return_resp_instance():组装UnifiedResponseModel统一响应体;http_exception():构造 FastAPIHTTPException;to_sse_event()/to_sse_event_instance()/to_sse_event_instance_str():将错误序列化为 SSE(Server-Sent Events)事件,适配流式输出场景;to_dict()/to_json_str():输出字典或 JSON 字符串;websocket_close_message():向 WebSocket 发送错误消息并可主动关闭连接。
这些方法表明 BISHENG 的异常体系同时服务于 HTTP、SSE 流式、WebSocket 等多种通信协议,是统一错误响应的基础设施。
4.2 业务异常定义示例
以 user.py 为例,用户模块的功能号段为106,每个具体错误占用后两位:
from .base import BaseErrorCode # Return error code related to user module, function module code:106 class UserValidateError(BaseErrorCode): Code: int = 10600 Msg: str = 'Account or password error' class UserPasswordExpireError(BaseErrorCode): Code: int = 10601 Msg: str = 'Your password has expired, please change it in time' class UserNameAlreadyExistError(BaseErrorCode): Code: int = 10605 Msg: str = 'User Name already exist' class UserPasswordStrengthError(BaseErrorCode): Code: int = 10622 Msg: str = 'Password must be at least 8 characters and include uppercase, lowercase, number, and symbol'4.3 使用方式
在业务逻辑中遇到错误情况时,应抛出对应的自定义异常(如raise UserNotFoundError()),而不是直接返回错误码或字符串;API 层统一捕获这些异常,根据异常的Code和Msg生成一致的错误响应。这种"抛异常-集中捕获-统一响应"的模式,让前端与调用方拿到结构完全一致的错误契约。
4.4 日志记录规范
- 在关键业务流程、异常捕获点及重要操作步骤中使用 Python 标准库
logging或loguru记录日志,确保日志清晰、结构化,并携带用户 ID、请求 ID 等上下文信息; - 日志级别合理使用:
DEBUG用于开发调试、INFO用于正常操作、WARNING用于潜在问题、ERROR用于错误事件、CRITICAL用于严重错误; - 日志统一使用英文,保持国际化与专业性,消息简洁明了。
从仓库看,loguru 已成为 BISHENG 后端的实际日志方案,在assistant.py、flow.py、audit_log.py等众多服务文件中均以from loguru import logger引入并使用。
5. 注释与文档规范
- 注释和文档统一使用英文,保持国际化与专业性;
- Docstrings:核心公共函数、类、复杂的业务方法必须包含文档字符串,说明功能、参数详情与返回类型——这一点在
BaseRepositoryImpl、BaseRepository的每个方法上都有落实,如"""Save Entity"""、"""Find a single entity"""; - 内联注释:对反直觉的实现或复杂逻辑,必须说明"为什么"这么写,而不仅是"做了什么"(如
BaseRepositoryImpl.find_one中# Apply Filter Criteria这类注释)。
6. 规范的实际应用方式
这份规范以 Agent Skill 的形式存在于仓库的 src/backend/.agents/skills/coding_guidelines/SKILL.md,其 frontmatter 中声明了name: 开发规范 (Coding Guidelines)与description: 规范当前 Python (FastAPI + SQLAlchemy) 项目的代码开发标准,涵盖编码风格、分层架构、数据库及异常处理等。这意味着在 AI 辅助开发或答疑时,Agent 会将这份规范作为判断代码质量与架构合理性的评价标准——包括代码生成、重构、修改以及日常答疑环节。开发者同样可以将其作为代码评审的检查清单:提交代码前逐项核对命名风格、分层归属、迁移文件、异常处理与日志覆盖情况,从而保证整个仓库数十个业务模块的工程质量底线。
【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考