Skip to content

代码风格 ​

📍 适用:所有 Python 代码(src/app/, tests/, scripts/) 给 Claude / Cursor / Cody 等 AI 生成代码时遵守


1. 工具链 ​

工具用途配置
Rufflint + format 一体pyproject.toml [tool.ruff]
mypy strict类型检查pyproject.toml [tool.mypy.ini_options]
pre-commitgit 钩子.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: false

1.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"]

新增抑制时:

  1. 加进 pyproject.toml,并在紧邻的注释里写明理由(写「因为 X 契约 / 技术限制」, 不要写「需要忽略」这种同义反复)
  2. 不要在源码里写 # 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: datetime

2.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 → ⚠️ 告警不阻塞
码规则
E1logging.getLogger(...) —— stdlib logger
E2extra={...} —— 字段被静默丢弃
E3event=... —— structlog 保留字,运行必炸
W1%s 位置参数(值藏在 event 串里、不可结构化查询)
W2level= / 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_casesms_code_repository.py
Python 类PascalCaseSmsCodeRepository
Python 函数snake_caseverify_and_consume
Python 常量UPPER_SNAKEMAX_ATTEMPTS = 3
测试文件test_*.pytest_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, abstractmethod

4.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 SmsCodeRepositoryPort

6. 异步 / 并发 ​

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 函数禁止 sync

6.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.py

7.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. 相关文档 ​