Appearance
代码风格
📍 适用:所有 Python 代码(src/app/, tests/, scripts/) 给 Claude / Cursor / Cody 等 AI 生成代码时遵守
1. 工具链
| 工具 | 用途 | 配置 |
|---|---|---|
| Ruff | lint + format 一体 | pyproject.toml [tool.ruff] |
| mypy strict | 类型检查 | pyproject.toml [tool.mypy.ini_options] |
| pre-commit | git 钩子 | .pre-commit-config.yaml |
1.1 pyproject.toml 关键配置
toml
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "W", "I", "N", "UP", "B", "A", "C4", "DTZ", "T20", "PT", "Q", "RET", "SIM", "TID", "ARG", "PTH", "ERA", "PL", "TRY", "RUF"]
ignore = ["E501", "B008", "TRY003"] # line-too-long, function-call-in-default, raise-vanilla-args
[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
addopts = "-v --tb=short --cov=src --cov-report=term-missing --cov-fail-under=80"1.2 pre-commit hooks
yaml
# .pre-commit-config.yaml
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.7.0
hooks:
- id: ruff
args: [--fix, --exit-non-zero-on-fix]
- id: ruff-format
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.13.0
hooks:
- id: mypy
additional_dependencies: [pydantic, sqlmodel]
args: [--strict]
- repo: local
hooks:
- id: check layers
name: check DDD layers
entry: uv run python scripts/check_layers.py
language: system
pass_filenames: false1.3 抑制 lint 的唯一合法方式:登记在 pyproject,不写内联 noqa
2026-09-20 把仓库积存的 339 条内联 # noqa 全部清零,抑制收敛到 pyproject.toml 的 [tool.ruff.lint.per-file-ignores] —— 每条都带理由注释。 (⚠️ 本次未新增 [tool.ruff.lint].ignore 任何条目:331 条抑制全部落在 per-file-ignores。ignore 列表是既有的、按规则普适豁免的那批。)
toml
[tool.ruff.lint.per-file-ignores]
# N815(119 处)—— 字段名**就是 V1 接口契约**的 camelCase(`planOption` / `groupId` …),
# 改 snake_case 是破坏性变更。正解是 Pydantic alias_generator(未做,另案)。
"src/app/*/interface/api/*_actions.py" = ["N815"]新增抑制时:
- 加进
pyproject.toml,并在紧邻的注释里写明理由(写「因为 X 契约 / 技术限制」, 不要写「需要忽略」这种同义反复) - 不要在源码里写
# noqa—— 零散的抑制散落在 40 个文件里, 没人能审、也没人敢删(本次清理前就处于这个状态)
⚠️ 为什么禁止内联 noqa:per-file-ignores 是可审计的清单(一处就能看全项目 在压制什么);内联 noqa 是不可审计的散点。339 条内联注释里, 331 条其实是正当抑制(契约 camelCase / 导入顺序 / Celery 副作用导入 / 有意 naive 时间), 但因为散落各处、没有理由,长期被当成"技术债"。
⚠️ 代价(必须知道):per-file-ignores 是文件级的 —— 例:*_actions.py 登记了 N815 后,这些文件里新增的非契约 camelCase 也不会被拦。故在 payload 文件里 写 camelCase 字段时需人工确认「这确实是契约字段」。
2. Python 代码风格
2.1 类型注解(强制)
python
# ✅ 推荐
async def execute(self, input: SmsSendInput, client_ip_subnet: str) -> None:
...
# ❌ 禁止
async def execute(self, input, ip): # 无类型注解
...2.2 Pydantic v2(BaseModel + model_config)
python
from pydantic import BaseModel, ConfigDict, Field
class SmsSendInput(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
phone: str = Field(min_length=11, max_length=11, pattern=r"^1[3-9]\d{9}$")
captcha_id: str = Field(min_length=8, max_length=64)
captcha_code: str = Field(min_length=1, max_length=8)
client_id: str = Field(default="default", max_length=32, pattern=r"^[A-Za-z0-9_-]+$")关键约定:
extra="forbid":拒绝未知字段(严格契约)frozen=True:入参不可变(避免 use case 内部意外修改)Field约束而非手动校验(schema 严格 vs 业务宽松)
2.3 SQLModel Model
python
from datetime import datetime
from sqlmodel import SQLModel, Field
class Captcha(SQLModel, table=True):
__tablename__ = "oauth_captcha"
id: str = Field(primary_key=True)
code_hash: str
generated_ip_subnet: str
generated_at: datetime
expires_at: datetime2.4 抽象端口
python
from abc import ABC, abstractmethod
class SmsCodeRepositoryPort(ABC):
@abstractmethod
async def save(self, sms_code: SmsCode) -> None: ...
@abstractmethod
async def verify_and_consume(
self,
phone: Phone,
client_id: ClientId,
input_code_hash: str,
) -> VerifyResult: ...2.5 异常处理(业务码优先)
python
# ✅ 推荐:抛 OAuthError
from src.app.oauth.domain.exceptions import OAuthError, OAuthErrorCode
if not ok:
raise OAuthError(OAuthErrorCode.SMS_CODE_WRONG, "短信验证码错误")
# ❌ 禁止:HTTPException
from fastapi import HTTPException
if not ok:
raise HTTPException(status_code=400, detail="验证码错误") # OATH-M-006 风险2.6 日志(structlog)
python
# ✅ 推荐
import structlog
logger = structlog.get_logger(__name__)
logger.info("sms_send.success", phone=phone, client_id=cid)
logger.warning("sms_send.rate_limited", phone=phone, layer="phone_60s")
# ❌ 禁止:stdlib logging
import logging
logger = logging.getLogger(__name__)
logger.info(f"sms sent to {phone}") # 拼接字符串,无结构化字段注意:
- 手机号日志自动脱敏(参见 Pydantic processor)
- 字段名用 snake_case(
client_id而非clientId) - 业务事件(
UserRegistered)也走 structlog(与 prometheus 业务事件联动) - 异常一律
error=f"{type(exc).__name__}: {exc}"—— structlog 的JSONRenderer不展开exc_info,只带error=文本才留得下
接口字段命名(HTTP API 契约 · 2026-10-01 修订拍板 D2)
读路径容忍 snake_case,写路径 camelCase(One Console Round 1 D7「新接口禁 snake」的修订版):
- 历史 V1/V2 契约(
planOption/groupId/isNewUser等 camelCase):逐字不动 - 新 dataclass 域(admin 探针/告警/审批等
asdict()直出):snake_case 合法—— 前端 TypeScript interface 以后端实际响应为准对齐(探针process_resident_mb先例); 不为此补投影层(Round 2 拍板 D2:投影层 ≈ 半会话换零收益) - 新写路径入参(POST/PATCH body):camelCase 为准(与既有写契约一致,如
alertId) - 判定权威:
docs/api-docs/v1/<domain>/契约页 +qm-console/src/service/api/实况
⚠️ 为什么这条是硬约定而不是风格偏好
main.py 只给 structlog 配了 JSONRenderer,没有给 stdlib 路径接 ProcessorFormatter。于是 stdlib logger 有两种静默故障(2026-09-20 实测):
| 写法 | 实际后果 |
|---|---|
logging.getLogger(...) | 输出 INFO:name:event 纯文本,混在 JSON 流里 → 下游按 JSON 解析会跳过这些行 |
logger.info("x", extra={"user_id": ...}) | extra 的键一个字都不会输出(默认 format string 不含它们)—— 只有 error= 那种「值已在消息串里」的才碰巧留下 |
两者都不报错、不告警,只有去翻日志才发现字段没了。
机器卡控
bash
python3 scripts/check_logging.py # pre-commit 已挂(check-logging)- E1/E2/E3 → 🔴 阻塞提交
- W1/W2 → ⚠️ 告警不阻塞
| 码 | 规则 |
|---|---|
| E1 | logging.getLogger(...) —— stdlib logger |
| E2 | extra={...} —— 字段被静默丢弃 |
| E3 | event=... —— structlog 保留字,运行必炸 |
| W1 | %s 位置参数(值藏在 event 串里、不可结构化查询) |
| W2 | level= / timestamp= / logger=(被 processor 覆写) |
⚠️ E3 是实测踩出来的:structlog 的 BoundLogger.info(self, event, **kw)首个位置参数就叫 event —— 再传 event=... → TypeError: got multiple values for argument 'event'。 (2026-09-20 转换 zepp webhook 的 str(ev)[:200] 时踩到,靠一条既有单测才抓到。) 字段名请改用 raw_event= 之类。
⚠️ %s 为什么只是告警:structlog 会插值 % 位置参数(实测确认,值不丢), 但值会藏在 event 字符串里 —— ELK/Loki 里搜不到 user_id:u000001。 紧急修补可先用,别长期停在这。
3. 命名规范
3.1 文件名
| 类型 | 命名 | 示例 |
|---|---|---|
| Python 模块 | snake_case | sms_code_repository.py |
| Python 类 | PascalCase | SmsCodeRepository |
| Python 函数 | snake_case | verify_and_consume |
| Python 常量 | UPPER_SNAKE | MAX_ATTEMPTS = 3 |
| 测试文件 | test_*.py | test_sms_login_usecase.py |
3.2 领域事件(PascalCase PastTense)
python
class UserRegistered(DomainEvent):
user_id: str
phone: str
occurred_at: datetime
class SmsCodeSent(DomainEvent):
phone: Phone
client_id: ClientId
occurred_at: datetime禁止:
- ❌
user.registered(点分命名) - ❌
UserRegister(非过去式) - ❌
USER_REGISTERED(常量命名)
3.3 错误码
- 字符串值(不是 int)
- 5 位数字:40101xxx(OAuth 业务码区段)
- Enum 名称 = 业务场景的 PascalCase
python
class OAuthErrorCode(str, Enum):
SMS_CODE_NOT_FOUND = "40101001" # 短信码不存在
SMS_CODE_WRONG = "40101003" # 短信码错误
CAPTCHA_WRONG = "40101003" # ⚠️ 同 code,msg 区分4. 注释与文档字符串
4.1 模块级 docstring(强制)
python
"""OAuth 短信码仓储端口契约。
定义 SmsCodeRepositoryPort 抽象接口,由 infrastructure 层实现
(RedisSmsCodeRepository 生产 / InMemorySmsCodeRepository 测试)。
"""
from abc import ABC, abstractmethod4.2 复杂逻辑必加注释
python
# ⚠️ 短信码「错」与「不存在」是两个不同 code(40101003 vs 40101001)
# UI 文案必须区分(错 → 可继续试;不存在 → 必须重发)
if result == VerifyResult.WRONG:
raise OAuthError(OAuthErrorCode.SMS_CODE_WRONG, "短信验证码错误")4.3 禁止的注释风格
- ❌ 行尾
# noqa(应直接修复) - ❌ 注释掉的代码块(git 已记录)
- ❌ ASCII art 流程图(用 Mermaid 写在 docs/)
5. Import 顺序
Ruff I 规则自动处理:
python
# 标准库
from abc import ABC, abstractmethod
from datetime import datetime
# 第三方
from fastapi import APIRouter
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
# 项目内(按层级)
from src.app.oauth.domain.entities.sms_code import SmsCode
from src.app.oauth.domain.exceptions import OAuthError, OAuthErrorCode
from src.app.oauth.application.ports.sms_code_repository import SmsCodeRepositoryPort6. 异步 / 并发
6.1 async/await 一致
python
# ✅ 所有 IO 函数 async
async def execute(self, input: SmsSendInput) -> None: ...
async def save(self, captcha: Captcha) -> None: ...
# ❌ 禁止混用 sync/async(除非有明确理由)
def execute(self, input): ... # IO 函数禁止 sync6.2 Lua 原子脚本注释
python
script = """
-- ⚠️ 关键:GET + 校验 + 标记 consumed 必须原子
-- 否则并发请求会导致双重消费
local data = redis.call('GET', KEYS[1])
...
"""7. 测试规范
7.1 测试文件命名
tests/
├── conftest.py # 全局 fixture
├── fixtures/
│ └── oauth_errors.py # 错误码 fixture
├── unit/
│ ├── domain/
│ │ └── test_sms_code_state_machine.py
│ └── application/
│ └── test_sms_login_usecase.py
├── integration/
│ └── infrastructure/
│ └── test_redis_sms_code_repository.py
└── e2e/
└── test_oauth_sms_login_e2e.py7.2 测试函数命名
python
async def test_wrong_sms_code_raises_oauth_error():
"""测试码错误时抛 OAuthError"""
...7.3 mock 策略
python
# ✅ Domain 纯函数测试:零 mock
def test_fresh_captcha_is_valid():
captcha = Captcha(...)
assert captcha.is_valid("192.168.1.0/24") is True
# ✅ Application 测试:mock 所有 port
async def test_new_user_registers():
sms_repo = AsyncMock(spec=SmsCodeRepositoryPort)
sms_repo.verify_and_consume.return_value = VerifyResult.MATCH
uc = SmsLoginUseCase(sms_repo, ...)
...
# ✅ Infrastructure 测试:真 Redis
@pytest.mark.integration
async def test_redis_crud(redis_client):
repo = RedisSmsCodeRepository(redis_client)
...7.4 覆盖率门槛
- 单测覆盖率 ≥ 80%(CI 卡控
--cov-fail-under=80) - Domain 层必须 100% 覆盖(纯函数、无理由漏)
- Application 层 ≥ 90%(业务逻辑核心)
8. 禁止的反模式
| 反模式 | 正确做法 |
|---|---|
from src.app.oauth.infrastructure... import in domain | 用 port 注入 |
raise HTTPException in use case | 用 raise OAuthError |
print() 调试 | 用 structlog.get_logger().debug() |
time.sleep() in test | 用 asyncio.sleep() |
dict/Any 代替 Pydantic schema | 用 BaseModel |
业务码忘记 OAuthErrorCode.X 而用 magic string | 用 enum |
| Lua 脚本里字符串拼接参数 | 用 KEYS/ARGV 传参 |
| 测试覆盖率低于 80% | CI 卡控 |
9. 相关文档
- 模块约定 — DDD 4 层目录、跨模块共享
- Internal Wiki → 错误码体系 — OATH-M-006 教训
- Internal Wiki → DDD 4 层 — 范式沉淀
- OAuth 实施总览 — OAuth 完整实施