Coverage for rfpy/jobs/offload.py: 100%

1import logging 

2 

3from sqlalchemy.orm import sessionmaker 

4 

5from rfpy.jobs.events.action import handle_event 

6from rfpy.model.audit.event import AuditEvent 

7from rfpy.utils import configure_rfpy, build_engine 

8from rfpy.mail import stub 

9from rfpy import conf 

10 

11IS_UWSGI = False 

12 

13try: # pragma: no cover 

14 import uwsgi # type: ignore # noqa: F401 

15 import uwsgidecorators 

16 IS_UWSGI = True 

17 print('\nRunning UWSGI') 

18except ImportError: 

19 print('\nNot Running UWSGI') 

20 from rfpy.jobs import uwsgi_stubs as uwsgidecorators 

21 

22 

23log = logging.getLogger(__name__) 

24 

25 

26@uwsgidecorators.spool 

27def spool_event_handler(arg_dict: dict): 

28 if conf.CONF is None: # pragma: no cover 

29 configure_rfpy() 

30 event_id = arg_dict.get('event_id', None) 

31 if event_id is None: 

32 log.error('spool_event_handler called without event_id') 

33 return 

34 engine = build_engine() 

35 Session = sessionmaker(engine) 

36 session = Session() 

37 evt = session.query(AuditEvent).get(event_id) 

38 

39 if evt is None: 

40 log.error('Unable to handle event, ID %s not found', event_id) 

41 return 

42 try: 

43 handle_event(evt, session, stub) 

44 except Exception: 

45 # Ensure function returns so UWSGI doesn't re-spool 

46 log.exception('Failure processing Event #%s', event_id)