docs(plan): 拆分 observe.py 导入,避免 Task 2 提交时 ruff F401

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
guke
2026-07-06 10:58:28 +08:00
parent 804de02188
commit 70c5c4cf08
@@ -161,6 +161,8 @@ Expected: FAIL —— `ModuleNotFoundError: app.core.observe` 或无 `record_eve
- [ ] **Step 3: 实现 `app/core/observe.py`(先只放队列部分)**
> 注意:本步只放队列相关代码。中间件用到的 `os`/`time`/`Match` 及 `_SKIP_PATHS`/`_UNMATCHED`/`_SERVICE` 常量放到 Task 3 一并加入——否则本步提交时 ruff 会报 F401 未用导入。
新建 `app/core/observe.py`
```python
@@ -172,23 +174,9 @@ Expected: FAIL —— `ModuleNotFoundError: app.core.observe` 或无 `record_eve
from __future__ import annotations
import asyncio
import logging
import os
import time
from starlette.routing import Match
from app.core.config import settings
logger = logging.getLogger("shagua.observe")
# 不采集的路径(纯噪音):健康检查。
_SKIP_PATHS = frozenset({"/health"})
# 未匹配路由(404/扫描器)归一到此,防维度爆炸。
_UNMATCHED = "__unmatched__"
# service 字段:与 logging.py 同源(LOG_SERVICE_NAME),默认 app-server。
_SERVICE = os.getenv("LOG_SERVICE_NAME", "app-server")
# 有界事件队列(懒创建,见 get_queue):首次取用时在运行中的 loop 里建,避免 import 期
# 无 loop 的边角问题;put_nowait/get_nowait 不需运行中的 loop → 可在无 loop 下测试。
_queue: "asyncio.Queue[dict] | None" = None
@@ -321,7 +309,42 @@ def test_middleware_noop_when_disabled(monkeypatch):
Run: `pytest tests/test_observe.py -q -k middleware`
Expected: FAIL —— `AttributeError: module 'app.core.observe' has no attribute 'RequestMetricsMiddleware'`
- [ ] **Step 3: 实现中间件(追加到 `app/core/observe.py` 末尾)**
- [ ] **Step 3a: `app/core/observe.py` 补中间件用的导入与常量**
把顶部 import 段从
```python
from __future__ import annotations
import asyncio
from app.core.config import settings
```
改成
```python
from __future__ import annotations
import asyncio
import os
import time
from starlette.routing import Match
from app.core.config import settings
# 不采集的路径(纯噪音):健康检查。
_SKIP_PATHS = frozenset({"/health"})
# 未匹配路由(404/扫描器)归一到此,防维度爆炸。
_UNMATCHED = "__unmatched__"
# service 字段:与 logging.py 同源(LOG_SERVICE_NAME),默认 app-server。
_SERVICE = os.getenv("LOG_SERVICE_NAME", "app-server")
```
`_queue` / `_dropped` / `get_queue` / `take_dropped` / `record_event` 保持不动。)
- [ ] **Step 3b: 实现中间件(追加到 `app/core/observe.py` 末尾)**
```python
def _resolve_route(scope) -> str: