Repository navigation
Expand file tree
/
Copy path_types.py
More file actions
122 lines (86 loc) · 3.24 KB
/
Copy path_types.py
File metadata and controls
122 lines (86 loc) · 3.24 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
from __future__ import annotations
from collections.abc import Callable, Coroutine
from enum import Enum
from typing import Annotated, Any, TypeVar
from pydantic import BaseModel, ConfigDict, Field
from crawlee._utils.docs import docs_group
from crawlee._utils.models import timedelta_secs
from crawlee._utils.system import CpuInfo, MemoryUsageInfo
@docs_group('Event data')
class Event(str, Enum):
"""Names of all possible events that can be emitted using an `EventManager`."""
# Core events
PERSIST_STATE = 'persistState'
SYSTEM_INFO = 'systemInfo'
MIGRATING = 'migrating'
ABORTING = 'aborting'
EXIT = 'exit'
# Session pool events
SESSION_RETIRED = 'sessionRetired'
# Browser pool events
BROWSER_LAUNCHED = 'browserLaunched'
BROWSER_RETIRED = 'browserRetired'
BROWSER_CLOSED = 'browserClosed'
PAGE_CREATED = 'pageCreated'
PAGE_CLOSED = 'pageClosed'
# State events
CRAWLER_STATUS = 'crawlerStatus'
@docs_group('Event data')
class EventPersistStateData(BaseModel):
"""Data for the persist state event."""
model_config = ConfigDict(validate_by_name=True, validate_by_alias=True)
is_migrating: Annotated[bool, Field(alias='isMigrating')]
@docs_group('Event data')
class EventSystemInfoData(BaseModel):
"""Data for the system info event."""
model_config = ConfigDict(validate_by_name=True, validate_by_alias=True)
cpu_info: Annotated[CpuInfo, Field(alias='cpuInfo')]
memory_info: Annotated[
MemoryUsageInfo,
Field(alias='memoryInfo'),
]
@docs_group('Event data')
class EventMigratingData(BaseModel):
"""Data for the migrating event."""
model_config = ConfigDict(validate_by_name=True, validate_by_alias=True)
# The remaining time in seconds before the migration is forced and the process is killed
# Optional because it's not present when the event handler is called manually
time_remaining: Annotated[timedelta_secs | None, Field(alias='timeRemainingSecs')] = None
@docs_group('Event data')
class EventAbortingData(BaseModel):
"""Data for the aborting event."""
model_config = ConfigDict(validate_by_name=True, validate_by_alias=True)
@docs_group('Event data')
class EventExitData(BaseModel):
"""Data for the exit event."""
model_config = ConfigDict(validate_by_name=True, validate_by_alias=True)
@docs_group('Event data')
class EventCrawlerStatusData(BaseModel):
"""Data for the crawler status event."""
model_config = ConfigDict(validate_by_name=True, validate_by_alias=True)
message: str
"""A message describing the current status of the crawler."""
crawler_id: int
"""The ID of the crawler that emitted the event."""
EventData = (
EventPersistStateData
| EventSystemInfoData
| EventMigratingData
| EventAbortingData
| EventExitData
| EventCrawlerStatusData
)
"""A helper type for all possible event payloads"""
WrappedListener = Callable[..., Coroutine[Any, Any, None]]
TEvent = TypeVar('TEvent')
EventListener = (
Callable[
[TEvent],
Coroutine[Any, Any, None] | None,
]
| Callable[
[],
Coroutine[Any, Any, None] | None,
]
)
"""An event listener function - it can be both sync and async and may accept zero or one argument."""