Add the nightly audit-record purge job
Python · Python · intermediate · greenfield
Adds the retention policy module and the nightly purge job that enforces it. The policy computes the oldest kept day from RETENTION_DAYS; the purge job selects everything older and deletes it. Ran it against a staging snapshot and the record counts dropped as expected.
An audit-logging service under a 90-day compliance retention rule. The purge job runs nightly; deletions are irreversible (no soft delete, no backup restore inside the audit store).
Requirements
- Compliance requires keeping audit records for the trailing 90 calendar days counting today; app/retention/policy.py is the single source of truth for what may be deleted, and the purge job must follow its contract exactly.
- select_purgeable(records, today) returns exactly the records the policy allows deleting — deleting a record the policy says to keep is a compliance violation, the worse failure direction.
- records is a list of dicts with 'id' (str) and 'day' (datetime.date, the record's calendar day); it may be empty.
- purge_expired(store, today) deletes the purgeable records via store.delete(id) and returns how many it deleted. store.all_records() returns the full record list.
Files touched
- app/retention/policy.py
- app/retention/purge.py
--- app/retention/policy.py +"""Data-retention policy for audit records. + +Compliance retains audit records for the trailing RETENTION_DAYS calendar +days, COUNTING TODAY: with RETENTION_DAYS = 90, records dated today or any +of the 89 days before it are inside the window and must be kept. +""" + +from datetime import date, timedelta + +RETENTION_DAYS = 90 + + +def oldest_kept_day(today: date) -> date: + """Return the oldest day that must still be KEPT (inclusive). + + Records dated on this day or later are inside the retention window and + must never be deleted. Only records dated STRICTLY BEFORE this day are + eligible for purging. + """ + return today - timedelta(days=RETENTION_DAYS - 1) + --- app/retention/purge.py +"""Nightly purge job: delete audit records that fell out of retention.""" + +from datetime import date + +from app.retention.policy import oldest_kept_day + +