Coverage for src/backend/InvenTree/machine/tasks.py: 53%

1"""Background task definitions for the 'machine' app.""" 

2 

3import structlog 

4from opentelemetry import trace 

5 

6from common.settings import get_global_setting 

7from InvenTree.tasks import ScheduledTask, scheduled_task 

8from machine import registry 

9 

10tracer = trace.get_tracer(__name__) 

11logger = structlog.get_logger('inventree') 

12 

13 

14@tracer.start_as_current_span('ping_machines') 

15@scheduled_task(ScheduledTask.MINUTES, 5) 

16def ping_machines(): 

17 """Periodically ping all configured machines to check if they are online.""" 

18 if not get_global_setting('MACHINE_PING_ENABLED', True): 

19 return 

20 

21 for driver in registry.get_drivers(): 

22 logger.debug("Pinging machines for driver '%s'", driver.SLUG) 

23 driver.ping_machines()