Coverage for src/backend/InvenTree/importer/tasks.py: 26%

1"""Task definitions for the 'importer' app.""" 

2 

3from datetime import timedelta 

4 

5import structlog 

6 

7import InvenTree.helpers 

8import InvenTree.tasks 

9 

10logger = structlog.get_logger('inventree') 

11 

12 

13def import_data(session_id: int): 

14 """Load data from the provided file. 

15 

16 Attempt to load data from the provided file, and potentially handle any errors. 

17 """ 

18 import importer.models 

19 

20 try: 

21 session = importer.models.DataImportSession.objects.get(pk=session_id) 

22 logger.info("Loading data from session ID '%s'", session_id) 

23 session.import_data() 

24 except (ValueError, importer.models.DataImportSession.DoesNotExist): 

25 logger.error("Data import session with ID '%s' does not exist", session_id) 

26 return 

27 

28 

29@InvenTree.tasks.scheduled_task(InvenTree.tasks.ScheduledTask.DAILY) 

30def cleanup_import_sessions(): 

31 """Periodically remove old import sessions. 

32 

33 Every 5 days, remove any importer sessions that are more than 5 days old 

34 """ 

35 CLEANUP_DAYS = 5 

36 

37 import importer.models 

38 

39 if not InvenTree.tasks.check_daily_holdoff('cleanup_import_sessions', CLEANUP_DAYS): 

40 return 

41 

42 logger.info('Cleaning old data import sessions') 

43 

44 before = InvenTree.helpers.current_date() - timedelta(days=CLEANUP_DAYS) 

45 

46 sessions = importer.models.DataImportSession.objects.filter(timestamp__lte=before) 

47 

48 if sessions.count() > 0: 

49 logger.info('Deleting %s old data import sessions', sessions.count()) 

50 sessions.delete() 

51 

52 InvenTree.tasks.record_task_success('cleanup_import_sessions')