Coverage for src / qdrant_loader / webhooks / single_event_handler.py: 79%

1"""Single-event webhook handler for WS-5 (Jira).""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from qdrant_loader.utils.logging import LoggingConfig 

8from qdrant_loader.webhooks.queue_backend import ( 

9 SINGLE_DELETE, 

10 SINGLE_UPSERT, 

11 ChangeEvent, 

12) 

13 

14logger = LoggingConfig.get_logger(__name__) 

15 

16 

17async def parse_webhook_event( 

18 source_type: str, 

19 source: str, 

20 payload: Any, 

21) -> ChangeEvent | None: 

22 """Parse webhook payload into a ChangeEvent for Jira single-event processing.""" 

23 if source_type != "jira": 

24 return None 

25 

26 if not isinstance(payload, dict): 

27 logger.warning("Jira payload is not a dict", payload_type=type(payload)) 

28 return None 

29 

30 try: 

31 issue = payload.get("issue", {}) 

32 issue_key = issue.get("key") 

33 webhook_event = payload.get("webhookEvent", "unknown") 

34 

35 if not issue_key: 

36 logger.warning("No issue.key in Jira webhook payload") 

37 return None 

38 

39 operation = _map_jira_event_to_operation(webhook_event) 

40 if operation is None: 

41 logger.debug( 

42 "Unsupported Jira webhook event type", 

43 event_type=webhook_event, 

44 ) 

45 return None 

46 

47 return ChangeEvent( 

48 source=source, 

49 source_type=source_type, 

50 project_id=None, 

51 entity_id=issue_key, 

52 payload=payload, 

53 operation=operation, 

54 ) 

55 except Exception as exc: 

56 logger.exception("Failed to parse Jira webhook event", exc_info=exc) 

57 return None 

58 

59 

60def _map_jira_event_to_operation(jira_event: str) -> str | None: 

61 """Map Jira webhook event type to WS-5 operation.""" 

62 mapping = { 

63 "jira:issue_created": SINGLE_UPSERT, 

64 "jira:issue_updated": SINGLE_UPSERT, 

65 "jira:issue_deleted": SINGLE_DELETE, 

66 } 

67 return mapping.get(jira_event)