Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/events/ordering/__init__.py: 78%

1""" 

2Manages the partial causal ordering of events for a particular consumer. This module 

3maintains a buffer of events to be processed, aiming to process them in the order they 

4occurred causally. 

5""" 

6 

7import abc 

8from datetime import timedelta 

9import importlib 

10from typing import ( 

11 TYPE_CHECKING, 

12 AsyncContextManager, 

13 List, 

14 Protocol, 

15 Union, 

16 runtime_checkable, 

17) 

18from uuid import UUID 

19from prefect.logging import get_logger 

20from prefect.server.events.schemas.events import Event, ReceivedEvent 

21from prefect.settings import get_current_settings 

22 

23if TYPE_CHECKING: 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 import logging 

25 

26logger: "logging.Logger" = get_logger(__name__) 

27 

28# How long we'll retain preceding events (to aid with ordering) 

29PRECEDING_EVENT_LOOKBACK = timedelta(minutes=15) 

30 

31# How long we'll retain events we've processed (to prevent re-processing an event) 

32PROCESSED_EVENT_LOOKBACK = timedelta(minutes=30) 

33 

34# How long we'll remember that we've seen an event 

35SEEN_EXPIRATION = max(PRECEDING_EVENT_LOOKBACK, PROCESSED_EVENT_LOOKBACK) 

36 

37# How deep we'll allow the recursion to go when processing events 

38MAX_DEPTH_OF_PRECEDING_EVENT = 20 

39 

40 

41@runtime_checkable 

42class CausalOrderingModule(Protocol): 

43 CausalOrdering: type["CausalOrdering"] 

44 

45 

46class EventArrivedEarly(Exception): 

47 def __init__(self, event: ReceivedEvent): 

48 self.event = event 

49 

50 

51class MaxDepthExceeded(Exception): 

52 def __init__(self, event: ReceivedEvent): 

53 self.event = event 

54 

55 

56class event_handler(Protocol): 

57 async def __call__( 57 ↛ exitline 57 didn't return from function '__call__' because

58 self, event: ReceivedEvent, depth: int = 0 

59 ) -> None: ... # pragma: no cover 

60 

61 

62class CausalOrdering(abc.ABC): 

63 def __init__(self, scope: str): 

64 self.scope = scope 

65 

66 @abc.abstractmethod 

67 async def event_has_been_seen(self, event: Union[UUID, Event]) -> bool: ... 67 ↛ exitline 67 didn't return from function 'event_has_been_seen' because

68 

69 @abc.abstractmethod 

70 async def record_event_as_seen(self, event: ReceivedEvent) -> None: ... 70 ↛ exitline 70 didn't return from function 'record_event_as_seen' because

71 

72 @abc.abstractmethod 

73 async def record_follower(self, event: ReceivedEvent) -> None: ... 73 ↛ exitline 73 didn't return from function 'record_follower' because

74 

75 @abc.abstractmethod 

76 async def forget_follower(self, follower: ReceivedEvent) -> None: ... 76 ↛ exitline 76 didn't return from function 'forget_follower' because

77 

78 @abc.abstractmethod 

79 async def get_followers(self, leader: ReceivedEvent) -> List[ReceivedEvent]: ... 79 ↛ exitline 79 didn't return from function 'get_followers' because

80 

81 @abc.abstractmethod 

82 async def get_lost_followers(self) -> List[ReceivedEvent]: ... 82 ↛ exitline 82 didn't return from function 'get_lost_followers' because

83 

84 @abc.abstractmethod 

85 def preceding_event_confirmed( 85 ↛ exitline 85 didn't return from function 'preceding_event_confirmed' because

86 self, handler: event_handler, event: ReceivedEvent, depth: int = 0 

87 ) -> AsyncContextManager[None]: ... 

88 

89 

90def get_triggers_causal_ordering() -> CausalOrdering: 

91 import_path = get_current_settings().server.events.causal_ordering 

92 causal_ordering_module = importlib.import_module(import_path) 

93 

94 if not isinstance(causal_ordering_module, CausalOrderingModule): 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 raise ValueError( 

96 f"Module at {import_path} does not export a CausalOrdering class. Please check your server.events.causal_ordering setting." 

97 ) 

98 

99 return causal_ordering_module.CausalOrdering(scope="triggers")