import logging import shutil from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any from src.config import get_settings logger = logging.getLogger(__name__) def cleanup_old_efs_directories(path: Path, retention_days: int) -> None: today_midnight = datetime.now(UTC).replace( hour=0, minute=0, second=0, microsecond=0 ) cutoff_timestamp = int( (today_midnight - timedelta(days=retention_days)).timestamp() ) errors: list[str] = [] for item in path.iterdir(): if not item.is_dir(): continue # Directory names are Unix timestamps if not item.name.isdecimal(): logger.error("Non-numeric EFS directory name: %s", item.name) errors.append(f"Non-numeric EFS directory name: {item.name}") continue dir_timestamp = int(item.name) if dir_timestamp < cutoff_timestamp: try: shutil.rmtree(item) logger.info("Deleted EFS directory %s", item.name) except Exception: logger.exception( "Failed to delete EFS directory %s", item.name, ) errors.append(f"Failed to delete EFS directory: {item.name}") if errors: raise RuntimeError( f"EFS cleanup encountered {len(errors)} error(s): " + "; ".join(errors) ) def handler(event: dict[str, Any] | None, context: Any) -> None: try: settings = get_settings() logging.getLogger().setLevel(settings.LOG_LEVEL) cleanup_old_efs_directories(settings.EFS_PATH, settings.EFS_RETENTION_DAYS) except Exception: logger.exception("Handler failed") raise