53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
import json
|
|
import logging
|
|
import threading
|
|
import time
|
|
from typing import Callable
|
|
|
|
from ..audit import AuditFn
|
|
from .policy import RetentionPolicy, compute_cutoffs
|
|
from .repository import RetentionRepository, RunRecord
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SYSTEM_USER = "system"
|
|
ORIGIN_AUTO = "auto"
|
|
ORIGIN_MANUAL = "manual"
|
|
AUDIT_RUN = "retention.run"
|
|
AUDIT_TARGET = "retention"
|
|
|
|
|
|
class RetentionBusy(Exception):
|
|
pass
|
|
|
|
|
|
class RetentionService:
|
|
def __init__(self, repo: RetentionRepository, policy_reader: Callable[[], RetentionPolicy], audit: AuditFn,
|
|
clock: Callable[[], float] = time.time):
|
|
self._repo = repo
|
|
self._policy = policy_reader
|
|
self._audit = audit
|
|
self._clock = clock
|
|
self._lock = threading.Lock()
|
|
|
|
def last_run(self) -> RunRecord | None:
|
|
return self._repo.last_run()
|
|
|
|
def run(self, origin: str = ORIGIN_AUTO, user: str = SYSTEM_USER) -> RunRecord:
|
|
if not self._lock.acquire(blocking=False):
|
|
raise RetentionBusy
|
|
try:
|
|
now = int(self._clock())
|
|
counts = self._repo.purge(compute_cutoffs(self._policy(), now))
|
|
try:
|
|
self._repo.checkpoint()
|
|
except Exception as exc:
|
|
log.warning("wal_checkpoint fehlgeschlagen: %s", type(exc).__name__)
|
|
record = RunRecord(now, origin, counts)
|
|
self._repo.save_run(record)
|
|
log.info("Aufbewahrung: gelöschte Zeilen %s", counts)
|
|
self._audit(user, AUDIT_RUN, AUDIT_TARGET, None, json.dumps(counts))
|
|
return record
|
|
finally:
|
|
self._lock.release()
|