Coverage for /usr/local/lib/python3.12/site-packages/prefect/client/schemas/events.py: 50%

1from typing import TYPE_CHECKING 1 ctx1a

2 

3from pydantic import AnyHttpUrl, Field 1 ctx1a

4 

5from prefect._internal.schemas.bases import PrefectBaseModel 1 ctx1a

6from prefect.events.schemas.events import ReceivedEvent 1 ctx1a

7 

8if TYPE_CHECKING: 8 ↛ 9line 8 didn't jump to line 9 because the condition on line 8 was never true1 ctx1a

9 from prefect.client.orchestration import PrefectClient, SyncPrefectClient 

10 

11 

12class EventPage(PrefectBaseModel): 1 ctx1a

13 """a single page of events returned from the API""" 

14 

15 events: list[ReceivedEvent] = Field(description="the events matching the query") 1 ctx1a

16 total: int = Field(description="the total number of matching events") 1 ctx1a

17 next_page: AnyHttpUrl | None = Field( 1 ctx1a

18 description="the URL for the next page of results, if there are more" 

19 ) 

20 

21 async def get_next_page(self, client: "PrefectClient") -> "EventPage | None": 1 ctx1a

22 """ 

23 fetch the next page of events. 

24 

25 args: 

26 client: the PrefectClient instance to use for fetching 

27 

28 returns: 

29 the next EventPage, or None if there are no more pages 

30 """ 

31 if not self.next_page: 

32 return None 

33 return await client.read_events_page(self.next_page) 

34 

35 def get_next_page_sync(self, client: "SyncPrefectClient") -> "EventPage | None": 1 ctx1a

36 """ 

37 fetch the next page of events (sync version). 

38 

39 args: 

40 client: the SyncPrefectClient instance to use for fetching 

41 

42 returns: 

43 the next EventPage, or None if there are no more pages 

44 """ 

45 if not self.next_page: 

46 return None 

47 return client.read_events_page(self.next_page)