freedak/Scrapling
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: Correcting return type annotation for Fetcher/AsyncFetcher (#93)

Browse Source
This commit is contained in:
Karim shoair
2025-10-01 04:17:59 +03:00
parent 7aa083053b
commit 849913ad77
2 changed files with 379 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scrapling/engines/static.py
+378
View File
@@ -650,6 +650,195 @@ class FetcherClient(FetcherSession):
self.__aexit__ = None
self._curl_session = True
# Setting the correct return types for the type checking/autocompletion
def get(
self,
url: str,
params: Optional[Dict | List | Tuple] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Response:
return super().get(
url,
params,
headers,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
def post(
self,
url: str,
data: Optional[Dict | str] = None,
json: Optional[Dict | List] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
params: Optional[Dict | List | Tuple] = None,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Response:
return super().post(
url,
data,
json,
headers,
params,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
def put(
self,
url: str,
data: Optional[Dict | str] = None,
json: Optional[Dict | List] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
params: Optional[Dict | List | Tuple] = None,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Response:
return super().put(
url,
data,
json,
headers,
params,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
def delete(
self,
url: str,
data: Optional[Dict | str] = None,
json: Optional[Dict | List] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
params: Optional[Dict | List | Tuple] = None,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Response:
return super().delete(
url,
data,
json,
headers,
params,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
class AsyncFetcherClient(FetcherSession):
def __init__(self, *args, **kwargs):
@@ -659,3 +848,192 @@ class AsyncFetcherClient(FetcherSession):
self.__aenter__ = None
self.__aexit__ = None
self._async_curl_session = True
# Setting the correct return types for the type checking/autocompletion
def get(
self,
url: str,
params: Optional[Dict | List | Tuple] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Awaitable[Response]:
return super().get(
url,
params,
headers,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
def post(
self,
url: str,
data: Optional[Dict | str] = None,
json: Optional[Dict | List] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
params: Optional[Dict | List | Tuple] = None,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Awaitable[Response]:
return super().post(
url,
data,
json,
headers,
params,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
def put(
self,
url: str,
data: Optional[Dict | str] = None,
json: Optional[Dict | List] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
params: Optional[Dict | List | Tuple] = None,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Awaitable[Response]:
return super().put(
url,
data,
json,
headers,
params,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
def delete(
self,
url: str,
data: Optional[Dict | str] = None,
json: Optional[Dict | List] = None,
headers: Optional[Mapping[str, Optional[str]]] = _UNSET,
params: Optional[Dict | List | Tuple] = None,
cookies: Optional[CookieTypes] = None,
timeout: Optional[int | float] = _UNSET,
follow_redirects: Optional[bool] = _UNSET,
max_redirects: Optional[int] = _UNSET,
retries: Optional[int] = _UNSET,
retry_delay: Optional[int] = _UNSET,
proxies: Optional[ProxySpec] = _UNSET,
proxy: Optional[str] = _UNSET,
proxy_auth: Optional[Tuple[str, str]] = _UNSET,
auth: Optional[Tuple[str, str]] = None,
verify: Optional[bool] = _UNSET,
cert: Optional[str | Tuple[str, str]] = _UNSET,
impersonate: Optional[BrowserTypeLiteral] = _UNSET,
http3: Optional[bool] = _UNSET,
stealthy_headers: Optional[bool] = _UNSET,
**kwargs,
) -> Awaitable[Response]:
return super().delete(
url,
data,
json,
headers,
params,
cookies,
timeout,
follow_redirects,
max_redirects,
retries,
retry_delay,
proxies,
proxy,
proxy_auth,
auth,
verify,
cert,
impersonate,
http3,
stealthy_headers,
**kwargs,
)
scrapling/fetchers/requests.py
+1 -9
View File
@@ -1,17 +1,9 @@
from scrapling.core._types import (
Callable,
Dict,
List,
Optional,
SelectorWaitStates,
Iterable,
)
from scrapling.engines.static import (
FetcherSession,
FetcherClient as _FetcherClient,
AsyncFetcherClient as _AsyncFetcherClient,
)
from scrapling.engines.toolbelt.custom import BaseFetcher, Response
from scrapling.engines.toolbelt.custom import BaseFetcher
__FetcherClientInstance__ = _FetcherClient()