61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
"""AwoooP Phase 2.4: Project ID Context Variable
|
||
================================================
|
||
2026-05-04 ogt + Claude Sonnet 4.6（ADR-123 background loop tagging）
|
||
|
||
設計原則：
|
||
- Python asyncio.create_task() 自動繼承父任務的 ContextVar 值
|
||
- 起始流程不再在 lifespan 強制寫入固定 PROJECT_ID；呼叫端需明確提供 project_id
|
||
- get_db_context() 僅接受明確參數或已注入的 contextvar 作為 tenant 來源
|
||
- 多租戶未來：呼叫端傳入不同 project_id 即可隔離，無需改 loop 本體
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from contextvars import ContextVar, Token
|
||
|
||
# 追蹤當前非同步任務的 project_id
|
||
# Fail-Closed: 移除 default="awoooi"，進 DB 路徑需要明確租戶標籤
|
||
PROJECT_ID: ContextVar[str | None] = ContextVar("project_id")
|
||
PROJECT_ID_SOURCE: ContextVar[str | None] = ContextVar("project_id_source")
|
||
PROJECT_ID_REQUEST_ID: ContextVar[str | None] = ContextVar("project_id_request_id")
|
||
|
||
|
||
def set_project_context(
|
||
project_id: str | None,
|
||
source: str = "runtime",
|
||
request_id: str | None = None,
|
||
) -> tuple[Token[str | None], Token[str | None], Token[str | None]]:
|
||
"""
|
||
設定當前 request/context 的 project 上下文，並回傳 ContextVar token 供 restore。
|
||
"""
|
||
return (
|
||
PROJECT_ID.set(project_id),
|
||
PROJECT_ID_SOURCE.set(source),
|
||
PROJECT_ID_REQUEST_ID.set(request_id),
|
||
)
|
||
|
||
|
||
def clear_project_context(tokens: tuple[Token[str | None], Token[str | None], Token[str | None]]) -> None:
|
||
"""清除 request 上下文，回復前一個 ContextVar 狀態。"""
|
||
PROJECT_ID_REQUEST_ID.reset(tokens[2])
|
||
PROJECT_ID_SOURCE.reset(tokens[1])
|
||
PROJECT_ID.reset(tokens[0])
|
||
|
||
|
||
def get_project_context() -> dict[str, str | None]:
|
||
"""取得目前上下文快照（可直接寫入 audit log）。"""
|
||
return {
|
||
"project_id": PROJECT_ID.get(None),
|
||
"source": PROJECT_ID_SOURCE.get(None),
|
||
"request_id": PROJECT_ID_REQUEST_ID.get(None),
|
||
}
|
||
|
||
|
||
def get_current_project_id() -> str | None:
|
||
"""取得當前任務的 project_id（給 service 層使用）"""
|
||
return PROJECT_ID.get(None)
|
||
|
||
|
||
def get_current_project_context() -> dict[str, str | None]:
|
||
"""取得可追溯上下文（同 get_project_context，保留 API 命名）。"""
|
||
return get_project_context()
|