offerpai_python_ai/.kiro/steering/代码开发风格文档.md

inclusion

inclusion
manual

代码开发风格文档

本项目为 FastAPI + SQLAlchemy (asyncio) 的 Python 3.12 后端项目，应用主目录为 app/。

项目结构

• app/config/ — 配置层：Pydantic Settings 统一配置
• app/core/ — 核心基础设施：数据库、Redis、鉴权、中间件、异常处理、日志、统一响应
• app/ai/ — AI 能力层：LLM 模型枚举与实例创建
• app/api/ — 路由层：REST API 接口定义
• app/models/ — ORM 模型层：SQLAlchemy 声明式映射
• app/services/ — 业务逻辑层：Service 类
• app/core/schemas/ — 公共 Schema：统一响应模型等

命名约定

文件命名

• 全部小写，下划线分隔，如 func_permission_service.py、user_func_usage_log.py
• 路由文件以业务名命名，如 health.py、resume.py
• ORM 模型文件与表名对应（去掉 bg_ 前缀），如 func_permission.py 对应 bg_func_permission

类命名

• Service 以 Service 结尾，如 FuncPermissionService
• ORM 模型用 PascalCase 业务名，无后缀，如 FuncPermission、UserFuncUsageLog
• Pydantic Schema 按用途命名：请求参数以 Param 结尾，响应以 Dto 结尾，如 ResumeParam、ResumeDto
• 枚举类以大写命名，如 LLM

变量与函数命名

• 函数和变量使用 snake_case，如 check_and_deduct、user_id
• 私有函数以单下划线开头，如 _insert_usage_log
• 常量使用全大写下划线，如 _FRIENDLY_MESSAGES、_SKIP_PATHS

类型注解

• 所有函数参数和返回值必须有类型注解
• ORM 模型字段使用 Mapped[T] + mapped_column() 声明
• Pydantic 模型字段使用标准类型注解 + Field()
• 可选字段使用 Optional[T] 或 T | None
• 集合类型使用 list[T]、dict[K, V]（Python 3.12 内置泛型）

注释规范

• 模块级注释使用文件顶部的 docstring，说明模块用途和使用示例
• 类注释使用 docstring，说明对应的表名和用途
• 方法注释使用 docstring，简洁描述功能
• 复杂逻辑用行内注释 # 说明

ORM 模型类注释

• 类 docstring 说明对应的表名和用途
• 特殊字段通过 comment 参数说明含义，如 comment="状态 1=启用 0=禁用"

Service 类注释

• 模块级 docstring 说明该服务的主要功能、依赖服务、使用的表
• 格式示例：

  """功能权限 Service
  
  校验用户功能权限并扣减库存，业务异常时回退。
  逻辑与 Java 端 FuncPermissionService 完全一致。
  """
  

• 每个方法用 docstring 简要说明逻辑流程，复杂方法可分步骤描述

分包规则

API 路由（app/api/）

• 每个业务模块一个路由文件，如 health.py、resume.py
• 使用 APIRouter(prefix="/xxx", tags=["xxx"]) 定义路由前缀和标签
• 在 app/main.py 中注册路由

Service（app/services/）

• 每个业务模块一个 Service 文件
• Service 类通过构造函数接收 AsyncSession，如 def __init__(self, session: AsyncSession)
• 不使用全局 Service 实例，每次请求通过依赖注入创建

ORM 模型（app/models/）

• 每个表一个模型文件
• 所有模型继承 app.core.database.Base
• 表名通过 __tablename__ 指定

Pydantic Schema（app/core/schemas/）

• 公共 Schema 放在 app/core/schemas/ 下，如 responses.py
• 业务相关的请求/响应 Schema 放在对应的 app/api/ 或 app/services/ 同级目录，或集中在 app/core/schemas/{功能模块}/ 下

获取当前登录用户

• 通过 RequestContext.user_id.get() 获取当前登录用户 ID
• 或通过依赖注入 Depends(require_login) 获取并校验
• 需要功能权限校验时使用 Depends(func_permission("func_code"))

接口规范

• Router 只负责参数接收和调用 Service，不写业务逻辑
• 白名单路径（无需鉴权）在 settings.auth_whitelist 中配置
• POST 用 @router.post()，GET 用 @router.get()
• 复杂参数使用 Pydantic 模型 + Body()，简单参数使用 Query() 或 Path()
• 路由方法直接返回业务数据，由 ResponseWrapMiddleware 自动包装为 StandardResponse

异常处理

• HTTP 异常使用 raise HTTPException(status_code=xxx, detail="描述")
• 简单断言直接使用 Python assert 或 if not ... raise
• 不要 catch 后吞掉异常，交由全局异常处理器（exceptions.py）统一处理
• 全局异常处理器已注册：HTTP异常、验证异常、断言异常、未知异常

Redis 使用规范

• 通过 app.core.redis.redis_client 或依赖注入 Depends(get_redis) 获取客户端
• key 命名与 Java 端保持一致，如 login:token:{userId}
• 值统一 JSON 序列化（json.dumps / json.loads）
• 设置 TTL 时使用 ex 参数（秒）

数据库设计风格

• 与 Java 端共享同一数据库，表结构由 Java 端管理
• 表名以 bg_ 前缀，下划线命名，如 bg_func_permission
• 主键 id，类型 BigInteger
• 时间字段使用 DateTime 类型，包含 create_time 和 update_time
• 逻辑删除字段 is_delete，类型 BigInteger，0=正常，非0=删除
• 状态字段用 Integer，0/1 表示，通过 comment 说明含义
• 查询使用 SQLAlchemy select() + where() 构建条件
• 更新使用 update() + where() + values()
• 会话通过 get_db() 依赖注入获取，自动 commit/rollback/close

异步规范

• 所有数据库操作、Redis 操作、HTTP 请求使用 async/await
• Service 方法统一使用 async def
• 路由处理函数统一使用 async def
• 避免在异步上下文中使用同步阻塞操作

AI 调用规范

• 通过 LLM 枚举创建模型实例：LLM.DEEPSEEK_V3.create(temperature=0)
• kwargs 透传给 LangChain ChatOpenAI（temperature、max_tokens 等）
• AI 调用应做好异常捕获和容错，单次失败不应影响整体流程
• 长耗时 AI 调用考虑异步执行

AI 输出 JSON 解析

• LLM 返回的 JSON 经常被 markdown 代码块（```json ... ```）包裹，禁止直接使用 LangChain 的 JsonOutputParser
• 统一使用 app.tool.json_helper.parse_llm_json 解析 AI 输出的 JSON 文本
• parse_llm_json 会自动剥离 markdown 代码块标记，并通过 json_repair 做容错修复
• 不要在各模块中自行编写 JSON 清洗/解析逻辑，统一复用 parse_llm_json

代码格式规范

紧凑风格

• 避免过度换行，保持代码紧凑易读
• 链式调用尽量写在一行，除非超过 120 字符
• 方法参数列表较多时，可适当换行但保持紧凑
• f-string 拼接优先写在一行

示例

推荐（紧凑风格）：

# 查询语句一行
result = await session.execute(select(FuncPermission).where(FuncPermission.func_code == func_code, FuncPermission.status == 1))

# 链式操作一行
perm = result.scalar_one_or_none()

# f-string 拼接一行
log.info(f"功能权限校验 userId:{user_id} funcCode:{func_code}")

# 方法参数紧凑排列
async def check_and_deduct(self, user_id: int, func_code: str) -> int:

# 多条件 where 紧凑排列
result = await self.session.execute(select(UserFuncPermissionStock).where(
    UserFuncPermissionStock.user_id == user_id, UserFuncPermissionStock.func_code == func_code))