Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/scheduler/tasks/purge_group_exports.py: 29%

1import datetime 1 ctx1a

2from pathlib import Path 1 ctx1a

3 

4from sqlalchemy import cast, select 1 ctx1a

5 

6from mealie.core import root_logger 1 ctx1a

7from mealie.core.config import get_app_dirs 1 ctx1a

8from mealie.db.db_setup import session_context 1 ctx1a

9from mealie.db.models._model_utils.datetime import NaiveDateTime 1 ctx1a

10from mealie.db.models.group.exports import GroupDataExportsModel 1 ctx1a

11 

12ONE_DAY_AS_MINUTES = 1440 1 ctx1a

13 

14 

15def purge_group_data_exports(max_minutes_old=ONE_DAY_AS_MINUTES): 1 ctx1a

16 """Purges all group exports after x days""" 

17 logger = root_logger.get_logger() 

18 

19 logger.debug("purging group data exports") 

20 limit = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=max_minutes_old) 

21 

22 with session_context() as session: 

23 stmt = select(GroupDataExportsModel).filter(cast(GroupDataExportsModel.expires, NaiveDateTime) <= limit) 

24 results = session.execute(stmt).scalars().all() 

25 

26 total_removed = 0 

27 for result in results: 

28 session.delete(result) 

29 Path(result.path).unlink(missing_ok=True) 

30 total_removed += 1 

31 

32 session.commit() 

33 

34 logger.info(f"finished purging group data exports. {total_removed} exports removed from group data") 

35 

36 

37def purge_excess_files() -> None: 1 ctx1a

38 """Purges all files in the uploads directory that are older than 2 days""" 

39 directories = get_app_dirs() 

40 logger = root_logger.get_logger() 

41 

42 limit = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=ONE_DAY_AS_MINUTES * 2) 

43 

44 for file in directories.GROUPS_DIR.glob("**/export/*.zip"): 

45 # TODO: fix comparison types 

46 if file.stat().st_mtime < limit: # type: ignore 

47 file.unlink() 

48 logger.debug(f"excess group file removed '{file}'") 

49 

50 logger.info("finished purging excess files")