Coverage for arclith / application / use_cases / purge.py: 100%

1from datetime import datetime, timedelta, timezone 

2from typing import Generic, TypeVar 

3 

4from arclith.domain.models.entity import Entity 

5from arclith.domain.ports.logger import Logger 

6from arclith.domain.ports.repository import Repository 

7 

8T = TypeVar("T", bound=Entity) 

9 

10 

11class PurgeUseCase(Generic[T]): 

12 def __init__(self, repository: Repository[T], logger: Logger, retention_days: float | None = None) -> None: 

13 self._repository = repository 

14 self._logger = logger 

15 self._retention_days = retention_days 

16 

17 async def execute(self) -> int: 

18 if self._retention_days is None: 

19 self._logger.info("⏭️ Purge skipped — retention is infinite") 

20 return 0 

21 

22 cutoff = datetime.now(timezone.utc) - timedelta(days=self._retention_days) 

23 expired = [ 

24 e for e in await self._repository.find_deleted() 

25 if e.deleted_at is not None and e.deleted_at <= cutoff 

26 ] 

27 

28 for entity in expired: 

29 await self._repository.delete(entity.uuid) 

30 

31 self._logger.info("🧹 Purge complete", purged=len(expired), retention_days=self._retention_days) 

32 return len(expired) 

33