-
Notifications
You must be signed in to change notification settings - Fork 26
Expand file tree
/
Copy path_utils.py
More file actions
147 lines (106 loc) · 4.33 KB
/
Copy path_utils.py
File metadata and controls
147 lines (106 loc) · 4.33 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
from __future__ import annotations
import asyncio
import builtins
import inspect
import sys
from collections.abc import Callable
from contextlib import asynccontextmanager
from functools import wraps
from importlib import metadata
from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast
if TYPE_CHECKING:
from collections.abc import AsyncIterator
T = TypeVar('T', bound=Callable[..., Any])
def ensure_context(attribute_name: str) -> Callable[[T], T]:
"""Create a decorator that ensures the context manager is initialized before executing the method.
The decorator checks if the calling instance has the specified attribute and verifies that it is set to `True`.
If the instance is inactive, it raises a `RuntimeError`. Works for both synchronous and asynchronous methods.
Args:
attribute_name: The name of the boolean attribute to check on the instance.
Returns:
A decorator that wraps methods with context checking.
"""
def decorator(method: T) -> T:
@wraps(method)
def sync_wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
if not getattr(self, attribute_name, False):
raise RuntimeError(f'The {self.__class__.__name__} is not active. Use it within the context.')
return method(self, *args, **kwargs)
@wraps(method)
async def async_wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
if not getattr(self, attribute_name, False):
raise RuntimeError(f'The {self.__class__.__name__} is not active. Use it within the async context.')
return await method(self, *args, **kwargs)
return cast('T', async_wrapper if inspect.iscoroutinefunction(method) else sync_wrapper)
return decorator
def get_system_info() -> dict:
python_version = '.'.join([str(x) for x in sys.version_info[:3]])
system_info: dict[str, str | bool] = {
'apify_sdk_version': metadata.version('apify'),
'apify_client_version': metadata.version('apify-client'),
'crawlee_version': metadata.version('crawlee'),
'python_version': python_version,
'os': sys.platform,
}
system_info['is_running_in_ipython'] = is_running_in_ipython()
return system_info
def is_running_in_ipython() -> bool:
return getattr(builtins, '__IPYTHON__', False)
# The order of the rendered API groups is defined in the website/docusaurus.config.js file.
GroupName = Literal[
'Actor',
'Charging',
'Configuration',
'Errors',
'Event data',
'Event managers',
'Events',
'Request loaders',
'Storage clients',
'Storage data',
'Storages',
]
def docs_group(group_name: GroupName) -> Callable[[T], T]: # noqa: ARG001
"""Mark a symbol for rendering and grouping in documentation.
This decorator is used solely for documentation purposes and does not modify the behavior
of the decorated callable.
Args:
group_name: The documentation group to which the symbol belongs.
Returns:
The original callable without modification.
"""
def wrapper(func: T) -> T:
return func
return wrapper
def docs_name(symbol_name: str) -> Callable[[T], T]: # noqa: ARG001
"""Rename a symbol for documentation rendering.
This decorator modifies only the displayed name of the symbol in the generated documentation
and does not affect its runtime behavior.
Args:
symbol_name: The name to be used in the documentation.
Returns:
The original callable without modification.
"""
def wrapper(func: T) -> T:
return func
return wrapper
class ReentrantLock:
"""A reentrant lock implementation for asyncio using asyncio.Lock."""
def __init__(self) -> None:
self._lock = asyncio.Lock()
self._owner: asyncio.Task | None = None
@asynccontextmanager
async def __call__(self) -> AsyncIterator[None]:
"""Acquire the lock if it's not already owned by the current task, otherwise proceed without acquiring."""
me = asyncio.current_task()
if me is None:
raise RuntimeError('ReentrantLock must be used within an asyncio.Task')
if self._owner is me:
yield
return
async with self._lock:
self._owner = me
try:
yield
finally:
self._owner = None