59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def load_history_for_config(config: dict[str, Any], ctx: dict[str, Any]) -> tuple[Any, list[dict[str, Any]]]:
|
|
resolve_history_path = ctx["resolve_history_path"]
|
|
load_history = ctx["load_history"]
|
|
history_lock = ctx["_HISTORY_LOCK"]
|
|
|
|
history_path = resolve_history_path(config)
|
|
with history_lock:
|
|
history = load_history(history_path)
|
|
return history_path, history
|
|
|
|
|
|
def append_generated_history(
|
|
history_path: Any,
|
|
generated_items: list[dict[str, Any]],
|
|
key_fields: list[str],
|
|
ctx: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
append_new_history = ctx["append_new_history"]
|
|
load_history = ctx["load_history"]
|
|
history_lock = ctx["_HISTORY_LOCK"]
|
|
|
|
with history_lock:
|
|
current_history = load_history(history_path)
|
|
items = generated_items if isinstance(generated_items, list) else []
|
|
return append_new_history(history_path, current_history, items, key_fields)
|
|
|
|
|
|
def upsert_generated_history(
|
|
history_path: Any,
|
|
generated_items: list[dict[str, Any]],
|
|
key_fields: list[str],
|
|
ctx: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
upsert_history_records = ctx["upsert_history_records"]
|
|
load_history = ctx["load_history"]
|
|
history_lock = ctx["_HISTORY_LOCK"]
|
|
|
|
with history_lock:
|
|
current_history = load_history(history_path)
|
|
items = generated_items if isinstance(generated_items, list) else []
|
|
return upsert_history_records(history_path, current_history, items, key_fields)
|
|
|
|
|
|
def clear_history_and_suppressions(config: dict[str, Any], ctx: dict[str, Any]) -> int:
|
|
resolve_history_path = ctx["resolve_history_path"]
|
|
save_history = ctx["save_history"]
|
|
clear_skip_suppressions = ctx["clear_skip_suppressions"]
|
|
history_lock = ctx["_HISTORY_LOCK"]
|
|
|
|
history_path = resolve_history_path(config)
|
|
with history_lock:
|
|
save_history(history_path, [])
|
|
return int(clear_skip_suppressions())
|