|
# retoor <retoor@molodetz.nl>
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from devplacepy.cache import TTLCache
|
|
from devplacepy.database import db, build_pagination
|
|
from devplacepy.services.audit import store
|
|
from devplacepy.services.audit.store import AUDIT_TABLE
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FILTERABLE = ("event_key", "category", "actor_role", "actor_uid", "origin", "result")
|
|
|
|
_options_cache = TTLCache(ttl=60, max_size=8)
|
|
|
|
|
|
def _build_where(filters: dict) -> tuple[str, dict]:
|
|
clauses: list[str] = []
|
|
params: dict = {}
|
|
for key in FILTERABLE:
|
|
value = filters.get(key)
|
|
if value:
|
|
clauses.append(f"{key} = :{key}")
|
|
params[key] = value
|
|
query_text = (filters.get("q") or "").strip()
|
|
if query_text:
|
|
clauses.append("(summary LIKE :q OR event_key LIKE :q OR target_label LIKE :q)")
|
|
params["q"] = f"%{query_text}%"
|
|
date_from = (filters.get("date_from") or "").strip()
|
|
if date_from:
|
|
clauses.append("created_at >= :date_from")
|
|
params["date_from"] = date_from
|
|
date_to = (filters.get("date_to") or "").strip()
|
|
if date_to:
|
|
clauses.append("created_at <= :date_to")
|
|
params["date_to"] = date_to + "~"
|
|
where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
|
|
return where, params
|
|
|
|
|
|
def list_events(filters: dict, page: int, per_page: int = 25) -> tuple[list, dict]:
|
|
if AUDIT_TABLE not in db.tables:
|
|
return [], build_pagination(page, 0, per_page)
|
|
where, params = _build_where(filters)
|
|
total_rows = list(
|
|
db.query(f"SELECT COUNT(*) AS c FROM {AUDIT_TABLE}{where}", **params)
|
|
)
|
|
total = int(total_rows[0]["c"]) if total_rows else 0
|
|
pagination = build_pagination(page, total, per_page)
|
|
offset = (pagination["page"] - 1) * pagination["per_page"]
|
|
rows = list(
|
|
db.query(
|
|
f"SELECT * FROM {AUDIT_TABLE}{where} "
|
|
f"ORDER BY created_at DESC LIMIT :limit OFFSET :offset",
|
|
limit=pagination["per_page"],
|
|
offset=offset,
|
|
**params,
|
|
)
|
|
)
|
|
return rows, pagination
|
|
|
|
|
|
def _distinct(column: str) -> list[str]:
|
|
if AUDIT_TABLE not in db.tables:
|
|
return []
|
|
rows = db.query(
|
|
f"SELECT DISTINCT {column} AS value FROM {AUDIT_TABLE} "
|
|
f"WHERE {column} IS NOT NULL AND {column} != '' ORDER BY {column}"
|
|
)
|
|
return [row["value"] for row in rows]
|
|
|
|
|
|
def filter_options() -> dict:
|
|
cached = _options_cache.get("options")
|
|
if cached is not None:
|
|
return cached
|
|
options = {
|
|
"event_key": _distinct("event_key"),
|
|
"category": _distinct("category"),
|
|
"actor_role": _distinct("actor_role"),
|
|
"origin": _distinct("origin"),
|
|
"result": _distinct("result"),
|
|
}
|
|
_options_cache.set("options", options)
|
|
return options
|
|
|
|
|
|
def get_event_with_links(uid: str) -> Optional[dict]:
|
|
event = store.get_event(uid)
|
|
if not event:
|
|
return None
|
|
return {"event": event, "links": store.get_links(uid)}
|