---
url: /docs/ai-contexts/code-style.md
---
# 代码风格

> 📍 适用：所有 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: 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** → ⚠️ 告警不阻塞

| 码 | 规则 |
|---|---|
| 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, 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. 相关文档

* [模块约定](/ai-contexts/module-conventions) — DDD 4 层目录、跨模块共享
* [Internal Wiki → 错误码体系](/internal-wiki/architecture/error-codes) — OATH-M-006 教训
* [Internal Wiki → DDD 4 层](/internal-wiki/architecture/ddd-layers) — 范式沉淀
* [OAuth 实施总览](/internal-wiki/architecture/oauth-implementation) — OAuth 完整实施