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
|
from __future__ import annotations
import inspect
import warnings
from functools import wraps
from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, overload
from twisted.internet.defer import Deferred, maybeDeferred
from twisted.internet.threads import deferToThread
from scrapy.exceptions import ScrapyDeprecationWarning
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Callable, Coroutine
_T = TypeVar("_T")
_P = ParamSpec("_P")
def deprecated(
use_instead: Any = None,
) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]:
"""This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted
when the function is used."""
def deco(func: Callable[_P, _T]) -> Callable[_P, _T]:
@wraps(func)
def wrapped(*args: _P.args, **kwargs: _P.kwargs) -> _T:
message = f"Call to deprecated function {func.__name__}."
if use_instead:
message += f" Use {use_instead} instead."
warnings.warn(message, category=ScrapyDeprecationWarning, stacklevel=2)
return func(*args, **kwargs)
return wrapped
if callable(use_instead):
deco = deco(use_instead)
use_instead = None
return deco
def defers(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]: # pragma: no cover
"""Decorator to make sure a function always returns a deferred"""
warnings.warn(
"@defers is deprecated, you can use maybeDeferred() directly if needed.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
@wraps(func)
def wrapped(*a: _P.args, **kw: _P.kwargs) -> Deferred[_T]:
return maybeDeferred(func, *a, **kw)
return wrapped
def inthread(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]:
"""Decorator to call a function in a thread and return a deferred with the
result
"""
@wraps(func)
def wrapped(*a: _P.args, **kw: _P.kwargs) -> Deferred[_T]:
return deferToThread(func, *a, **kw)
return wrapped
@overload
def _warn_spider_arg(
func: Callable[_P, Coroutine[Any, Any, _T]],
) -> Callable[_P, Coroutine[Any, Any, _T]]: ...
@overload
def _warn_spider_arg(
func: Callable[_P, AsyncGenerator[_T]],
) -> Callable[_P, AsyncGenerator[_T]]: ...
@overload
def _warn_spider_arg(func: Callable[_P, _T]) -> Callable[_P, _T]: ...
def _warn_spider_arg(
func: Callable[_P, _T],
) -> (
Callable[_P, _T]
| Callable[_P, Coroutine[Any, Any, _T]]
| Callable[_P, AsyncGenerator[_T]]
):
"""Decorator to warn if a ``spider`` argument is passed to a function."""
sig = inspect.signature(func)
def check_args(*args: _P.args, **kwargs: _P.kwargs) -> None:
bound = sig.bind(*args, **kwargs)
if "spider" in bound.arguments:
warnings.warn(
f"Passing a 'spider' argument to {func.__qualname__}() is deprecated and "
"the argument will be removed in a future Scrapy version.",
category=ScrapyDeprecationWarning,
stacklevel=3,
)
if inspect.iscoroutinefunction(func):
@wraps(func)
async def async_inner(*args: _P.args, **kwargs: _P.kwargs) -> _T:
check_args(*args, **kwargs)
return await func(*args, **kwargs)
return async_inner
if inspect.isasyncgenfunction(func):
@wraps(func)
async def asyncgen_inner(
*args: _P.args, **kwargs: _P.kwargs
) -> AsyncGenerator[_T]:
check_args(*args, **kwargs)
async for item in func(*args, **kwargs):
yield item
return asyncgen_inner
@wraps(func)
def sync_inner(*args: _P.args, **kwargs: _P.kwargs) -> _T:
check_args(*args, **kwargs)
return func(*args, **kwargs)
return sync_inner
|
