shangfangjian
317a5e338a
单文件 852 行按职责拆分,保持 BzzoiroSource 与 get_source("bzzoiro") 行为不变:
- bzzoiro_common HTTP 抓取(多 key 轮换) + 字段转换原语
- bzzoiro_events fetch_bzzoiro_events + BzzoiroSource.ingest + Bronze 补写
- bzzoiro_standings standings 管线
- bzzoiro_stats stats 回填
- pipeline_write RawEvent/IngestFailure/DataLineage 写入助手
子模块运行期经聚合门面 src.data.bzzoiro 解析可替换协作者,
单文件时代的 bz.* monkeypatch 语义完全保留。
路由 import 已指向新模块(ingest.py / schedules.py)。
81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
"""管线基础设施写入助手:RawEvent(Bronze 原始载荷)/ IngestFailure(死信)/ DataLineage(血缘)。
|
|
|
|
从 bzzoiro.py 拆出。约定(与拆分前一致):
|
|
- 只 add 不 commit —— 事务由调用方 UnitOfWork 控制,分批事务约定不变;
|
|
- 死信与 Bronze 写入同为 best-effort:失败只记 warning,绝不拖垮采集主流程。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from src.db.models import DataLineage, IngestFailure, RawEvent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload: dict, batch_id: str | None = None) -> None:
|
|
"""写入 Bronze 层原始事件(幂等:同 source_record_id 跳过)。"""
|
|
from sqlalchemy import select as _select
|
|
stmt = _select(RawEvent).where(
|
|
RawEvent.source_system == source_system,
|
|
RawEvent.source_record_id == source_record_id,
|
|
)
|
|
existing = (await db.execute(stmt)).scalar_one_or_none()
|
|
if existing is None:
|
|
db.add(RawEvent(
|
|
source_system=source_system,
|
|
source_record_id=source_record_id,
|
|
raw_payload=raw_payload,
|
|
ingest_batch_id=batch_id,
|
|
))
|
|
|
|
|
|
async def _write_ingest_failure(db, source_system: str, entity_type: str, source_record_id: str | None, error_type: str, error_detail: str | None, raw_payload: dict | None = None) -> None:
|
|
"""写入采集失败死信。"""
|
|
db.add(IngestFailure(
|
|
source_system=source_system,
|
|
entity_type=entity_type,
|
|
source_record_id=source_record_id,
|
|
error_type=error_type,
|
|
error_detail=error_detail,
|
|
raw_payload=raw_payload,
|
|
))
|
|
|
|
|
|
async def _safe_write_ingest_failure(
|
|
db,
|
|
*,
|
|
entity_type: str,
|
|
source_record_id: str | None,
|
|
error: Exception,
|
|
raw_payload: dict | None = None,
|
|
) -> None:
|
|
"""抓取失败时尽力写入死信表(失败不影响主流程)。
|
|
|
|
死信是「可观测性」基础设施,与 RawEvent/Lineage 同级:写入失败只记
|
|
warning,绝不能让原始抓取错误之外的新异常打断采集循环。
|
|
"""
|
|
try:
|
|
await _write_ingest_failure(
|
|
db, "bzzoiro", entity_type, source_record_id,
|
|
"fetch_error", str(error), raw_payload,
|
|
)
|
|
except Exception:
|
|
logger.warning(
|
|
"写入 ingest_failures 死信失败(entity=%s, record=%s): %s",
|
|
entity_type, source_record_id, error, exc_info=True,
|
|
)
|
|
|
|
|
|
async def _write_lineage(db, source_system: str, source_record_id: str, target_table: str, target_id: int | None, transform_name: str, transform_detail: dict | None = None, batch_id: str | None = None) -> None:
|
|
"""写入 ETL 血缘追踪。"""
|
|
db.add(DataLineage(
|
|
source_system=source_system,
|
|
source_record_id=source_record_id,
|
|
target_table=target_table,
|
|
target_id=target_id,
|
|
transform_name=transform_name,
|
|
transform_detail=transform_detail,
|
|
batch_id=batch_id,
|
|
))
|