71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""运行记录追踪器"""
|
|
import json
|
|
from datetime import datetime
|
|
|
|
class RunTracker:
|
|
"""ETL运行记录管理"""
|
|
|
|
def __init__(self, db_connection):
|
|
self.db = db_connection
|
|
|
|
def create_run(self, task_id: int, store_id: int, run_uuid: str,
|
|
export_dir: str, log_path: str, status: str,
|
|
window_start: datetime = None, window_end: datetime = None,
|
|
window_minutes: int = None, overlap_seconds: int = None,
|
|
request_params: dict = None) -> int:
|
|
"""创建运行记录"""
|
|
sql = """
|
|
INSERT INTO etl_admin.etl_run(
|
|
run_uuid, task_id, store_id, status, started_at, window_start, window_end,
|
|
window_minutes, overlap_seconds, fetched_count, loaded_count, updated_count,
|
|
skipped_count, error_count, unknown_fields, export_dir, log_path,
|
|
request_params, manifest, error_message, extra
|
|
) VALUES (
|
|
%s, %s, %s, %s, now(), %s, %s, %s, %s, 0, 0, 0, 0, 0, 0, %s, %s, %s,
|
|
'{}'::jsonb, NULL, '{}'::jsonb
|
|
)
|
|
RETURNING run_id
|
|
"""
|
|
|
|
result = self.db.query(
|
|
sql,
|
|
(run_uuid, task_id, store_id, status, window_start, window_end,
|
|
window_minutes, overlap_seconds, export_dir, log_path,
|
|
json.dumps(request_params or {}, ensure_ascii=False))
|
|
)
|
|
|
|
run_id = result[0]["run_id"]
|
|
self.db.commit()
|
|
return run_id
|
|
|
|
def update_run(self, run_id: int, counts: dict, status: str,
|
|
ended_at: datetime = None, manifest: dict = None,
|
|
error_message: str = None):
|
|
"""更新运行记录"""
|
|
sql = """
|
|
UPDATE etl_admin.etl_run
|
|
SET fetched_count = %s,
|
|
loaded_count = %s,
|
|
updated_count = %s,
|
|
skipped_count = %s,
|
|
error_count = %s,
|
|
unknown_fields = %s,
|
|
status = %s,
|
|
ended_at = %s,
|
|
manifest = %s,
|
|
error_message = %s
|
|
WHERE run_id = %s
|
|
"""
|
|
|
|
self.db.execute(
|
|
sql,
|
|
(counts.get("fetched", 0), counts.get("inserted", 0),
|
|
counts.get("updated", 0), counts.get("skipped", 0),
|
|
counts.get("errors", 0), counts.get("unknown_fields", 0),
|
|
status, ended_at,
|
|
json.dumps(manifest or {}, ensure_ascii=False),
|
|
error_message, run_id)
|
|
)
|
|
self.db.commit()
|