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 # type: ignore 

16 

17 IS_UWSGI = True 

18 print("\nRunning UWSGI") 

19except ImportError: 

20 print("\nNot Running UWSGI") 

21 from rfpy.jobs import uwsgi_stubs as uwsgidecorators 

22 

23 

24log = logging.getLogger(__name__) 

25 

26 

27@uwsgidecorators.spool 

28def spool_event_handler(arg_dict: dict): 

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

30 configure_rfpy() 

31 event_id = arg_dict.get("event_id", None) 

32 if event_id is None: 

33 log.error("spool_event_handler called without event_id") 

34 return 

35 engine = build_engine() 

36 Session = sessionmaker(engine, future=True) 

37 session = Session() 

38 evt = session.get(AuditEvent, event_id) 

39 

40 if evt is None: 

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

42 return 

43 try: 

44 handle_event(evt, session, stub) 

45 except Exception: 

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

47 log.exception("Failure processing Event #%s", event_id)