Coverage for postrfp/shared/fetch/audq.py: 100%

1""" 

2Audit event related queries 

3""" 

4 

5import logging 

6from typing import Optional 

7from sqlalchemy.orm import Query, Session, object_session 

8 

9from postrfp.model import ( 

10 AuditEvent, 

11 Organisation, 

12 Project, 

13) 

14from postrfp.model.audit.event import EventOrgACL 

15from postrfp.model.notify import WebhookSubscription 

16 

17log = logging.getLogger(__name__) 

18 

19 

20def audit_events(organisation: Organisation, event_type: Optional[str] = None) -> Query: 

21 eq = organisation.visible_events.order_by(AuditEvent.id.desc()) 

22 

23 if event_type: 

24 return eq.filter(AuditEvent.event_type == event_type) 

25 else: 

26 return eq 

27 

28 

29def project_audit_events( 

30 organisation: Organisation, project: Project, event_type: Optional[str] = None 

31) -> Query: 

32 ev_query = audit_events(organisation, event_type=event_type) 

33 return ev_query.filter(AuditEvent.project == project) 

34 

35 

36def latest_event(session: Session) -> Optional[AuditEvent]: 

37 assert session is not None 

38 return session.query(AuditEvent).order_by(AuditEvent.id.desc()).first() 

39 

40 

41def webhooks_for_event(event: AuditEvent) -> Query[WebhookSubscription]: 

42 session = object_session(event) 

43 assert session is not None 

44 q = ( 

45 session.query(WebhookSubscription) 

46 .join(Organisation) 

47 .join(EventOrgACL) 

48 .filter(EventOrgACL.event_id == event.id) 

49 .filter(WebhookSubscription.event_type == event.event_type) 

50 ) 

51 return q