131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.ext.asyncio import AsyncEngine
|
|
|
|
from app.core.time import iso_utc_ms
|
|
from app.db.session import create_sessionmaker, with_sqlite_busy_retry
|
|
|
|
DEFAULT_REQUEST_LOGS_KEEP_DAYS = 30
|
|
DEFAULT_REQUEST_LOGS_MAX_DELETE_ROWS = 50_000
|
|
DEFAULT_REQUEST_LOGS_CHUNK_SIZE = 1_000
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RequestLogsCleanupResult:
|
|
cutoff: str
|
|
deleted: int
|
|
has_more: bool
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RequestLogsCleanupPreview:
|
|
cutoff: str
|
|
would_delete: int
|
|
has_more: bool
|
|
|
|
|
|
def _clamp_int(value: int, *, min_v: int, max_v: int) -> int:
|
|
return max(min_v, min(int(value), max_v))
|
|
|
|
|
|
def _cutoff_iso(*, keep_days: int, now: datetime | None = None) -> str:
|
|
keep_days_i = int(keep_days)
|
|
if keep_days_i < 0:
|
|
raise ValueError("keep_days must be >= 0")
|
|
|
|
now_dt = now or datetime.now(timezone.utc)
|
|
cutoff_dt = now_dt - timedelta(days=keep_days_i)
|
|
return iso_utc_ms(cutoff_dt)
|
|
|
|
|
|
async def preview_request_logs_cleanup(
|
|
engine: AsyncEngine,
|
|
*,
|
|
keep_days: int = DEFAULT_REQUEST_LOGS_KEEP_DAYS,
|
|
max_delete_rows: int = DEFAULT_REQUEST_LOGS_MAX_DELETE_ROWS,
|
|
now: datetime | None = None,
|
|
) -> RequestLogsCleanupPreview:
|
|
cutoff = _cutoff_iso(keep_days=int(keep_days), now=now)
|
|
max_delete_rows_i = _clamp_int(int(max_delete_rows), min_v=1, max_v=10_000_000)
|
|
|
|
Session = create_sessionmaker(engine)
|
|
sql = """
|
|
SELECT id
|
|
FROM request_logs
|
|
WHERE created_at < :cutoff
|
|
ORDER BY created_at ASC
|
|
LIMIT :limit;
|
|
""".strip()
|
|
|
|
async def _op() -> RequestLogsCleanupPreview:
|
|
async with Session() as session:
|
|
res = await session.execute(sa.text(sql), {"cutoff": cutoff, "limit": int(max_delete_rows_i) + 1})
|
|
rows = res.all()
|
|
would_delete = min(len(rows), int(max_delete_rows_i))
|
|
has_more = len(rows) > int(max_delete_rows_i)
|
|
return RequestLogsCleanupPreview(cutoff=cutoff, would_delete=would_delete, has_more=has_more)
|
|
|
|
return await with_sqlite_busy_retry(_op)
|
|
|
|
|
|
async def cleanup_request_logs(
|
|
engine: AsyncEngine,
|
|
*,
|
|
keep_days: int = DEFAULT_REQUEST_LOGS_KEEP_DAYS,
|
|
max_delete_rows: int = DEFAULT_REQUEST_LOGS_MAX_DELETE_ROWS,
|
|
chunk_size: int = DEFAULT_REQUEST_LOGS_CHUNK_SIZE,
|
|
now: datetime | None = None,
|
|
) -> RequestLogsCleanupResult:
|
|
cutoff = _cutoff_iso(keep_days=int(keep_days), now=now)
|
|
max_delete_rows_i = _clamp_int(int(max_delete_rows), min_v=1, max_v=10_000_000)
|
|
chunk_size_i = _clamp_int(int(chunk_size), min_v=1, max_v=100_000)
|
|
chunk_size_i = min(int(chunk_size_i), int(max_delete_rows_i))
|
|
|
|
Session = create_sessionmaker(engine)
|
|
|
|
delete_sql = """
|
|
DELETE FROM request_logs
|
|
WHERE id IN (
|
|
SELECT id
|
|
FROM request_logs
|
|
WHERE created_at < :cutoff
|
|
ORDER BY created_at ASC
|
|
LIMIT :chunk
|
|
);
|
|
""".strip()
|
|
|
|
has_more_sql = """
|
|
SELECT 1
|
|
FROM request_logs
|
|
WHERE created_at < :cutoff
|
|
LIMIT 1;
|
|
""".strip()
|
|
|
|
async def _op() -> RequestLogsCleanupResult:
|
|
deleted = 0
|
|
has_more = False
|
|
|
|
async with Session() as session:
|
|
while deleted < int(max_delete_rows_i):
|
|
chunk = min(int(chunk_size_i), int(max_delete_rows_i) - int(deleted))
|
|
result = await session.execute(sa.text(delete_sql), {"cutoff": cutoff, "chunk": int(chunk)})
|
|
n = int(result.rowcount or 0)
|
|
if n <= 0:
|
|
await session.commit()
|
|
break
|
|
deleted += n
|
|
await session.commit()
|
|
|
|
if deleted >= int(max_delete_rows_i):
|
|
res = await session.execute(sa.text(has_more_sql), {"cutoff": cutoff})
|
|
has_more = res.first() is not None
|
|
|
|
return RequestLogsCleanupResult(cutoff=cutoff, deleted=int(deleted), has_more=bool(has_more))
|
|
|
|
return await with_sqlite_busy_retry(_op)
|
|
|