Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Typing: improve #92

Merged
merged 1 commit into from
Feb 8, 2025
Merged

Typing: improve #92

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 23 additions & 58 deletions primp/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,53 +3,18 @@
import asyncio
import sys
from functools import partial
from typing import TYPE_CHECKING, Literal, TypedDict
from typing import TYPE_CHECKING, TypedDict

if sys.version_info <= (3, 11):
from typing_extensions import Unpack
else:
from typing import Unpack

from .primp import RClient # type: ignore

if TYPE_CHECKING:
HttpMethod = Literal["GET", "HEAD", "OPTIONS", "DELETE", "POST", "PUT", "PATCH"]
IMPERSONATE = Literal[
"chrome_100", "chrome_101", "chrome_104", "chrome_105", "chrome_106",
"chrome_107", "chrome_108", "chrome_109", "chrome_114", "chrome_116",
"chrome_117", "chrome_118", "chrome_119", "chrome_120", "chrome_123",
"chrome_124", "chrome_126", "chrome_127", "chrome_128", "chrome_129",
"chrome_130", "chrome_131",
"safari_15.3", "safari_15.5", "safari_15.6.1", "safari_16",
"safari_16.5", "safari_17.0", "safari_17.2.1", "safari_17.4.1",
"safari_17.5", "safari_18", "safari_18.2",
"safari_ios_16.5", "safari_ios_17.2", "safari_ios_17.4.1", "safari_ios_18.1.1",
"safari_ipad_18",
"okhttp_3.9", "okhttp_3.11", "okhttp_3.13", "okhttp_3.14", "okhttp_4.9",
"okhttp_4.10", "okhttp_5",
"edge_101", "edge_122", "edge_127", "edge_131",
"firefox_109", "firefox_117", "firefox_128", "firefox_133",
] # fmt: skip
IMPERSONATE_OS = Literal["android", "ios", "linux", "macos", "windows"]

class RequestParams(TypedDict, total=False):
auth: tuple[str, str | None] | None
auth_bearer: str | None
params: dict[str, str] | None
headers: dict[str, str] | None
cookies: dict[str, str] | None
timeout: float | None
content: bytes | None
data: dict[str, str] | None
json: dict[str, str] | None
files: dict[str, str] | None

class ClientRequestParams(RequestParams):
impersonate: IMPERSONATE | None
impersonate_os: IMPERSONATE_OS | None
verify: bool | None
ca_cert_file: str | None
from .primp import RClient

if TYPE_CHECKING:
from .primp import IMPERSONATE, IMPERSONATE_OS, ClientRequestParams, HttpMethod, RequestParams, Response
else:

class _Unpack:
Expand Down Expand Up @@ -122,42 +87,42 @@ def __init__(
"""
super().__init__()

def __enter__(self):
def __enter__(self) -> Client:
return self

def __exit__(self, *args):
del self

def request(self, method: HttpMethod, url: str, **kwargs: Unpack[RequestParams]):
def request(self, method: HttpMethod, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return super().request(method=method, url=url, **kwargs)

def get(self, url: str, **kwargs: Unpack[RequestParams]):
def get(self, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return self.request(method="GET", url=url, **kwargs)

def head(self, url: str, **kwargs: Unpack[RequestParams]):
def head(self, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return self.request(method="HEAD", url=url, **kwargs)

def options(self, url: str, **kwargs: Unpack[RequestParams]):
def options(self, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return self.request(method="OPTIONS", url=url, **kwargs)

def delete(self, url: str, **kwargs: Unpack[RequestParams]):
def delete(self, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return self.request(method="DELETE", url=url, **kwargs)

def post(self, url: str, **kwargs: Unpack[RequestParams]):
def post(self, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return self.request(method="POST", url=url, **kwargs)

def put(self, url: str, **kwargs: Unpack[RequestParams]):
def put(self, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return self.request(method="PUT", url=url, **kwargs)

def patch(self, url: str, **kwargs: Unpack[RequestParams]):
def patch(self, url: str, **kwargs: Unpack[RequestParams]) -> Response:
return self.request(method="PATCH", url=url, **kwargs)


class AsyncClient(Client):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)

async def __aenter__(self):
async def __aenter__(self) -> AsyncClient:
return self

async def __aexit__(self, *args):
Expand All @@ -167,28 +132,28 @@ async def _run_sync_asyncio(self, fn, *args, **kwargs):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, partial(fn, *args, **kwargs))

async def request(self, method: HttpMethod, url: str, **kwargs: Unpack[RequestParams]):
async def request(self, method: HttpMethod, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self._run_sync_asyncio(super().request, method=method, url=url, **kwargs)

async def get(self, url: str, **kwargs: Unpack[RequestParams]):
async def get(self, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self.request(method="GET", url=url, **kwargs)

async def head(self, url: str, **kwargs: Unpack[RequestParams]):
async def head(self, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self.request(method="HEAD", url=url, **kwargs)

async def options(self, url: str, **kwargs: Unpack[RequestParams]):
async def options(self, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self.request(method="OPTIONS", url=url, **kwargs)

async def delete(self, url: str, **kwargs: Unpack[RequestParams]):
async def delete(self, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self.request(method="DELETE", url=url, **kwargs)

async def post(self, url: str, **kwargs: Unpack[RequestParams]):
async def post(self, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self.request(method="POST", url=url, **kwargs)

async def put(self, url: str, **kwargs: Unpack[RequestParams]):
async def put(self, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self.request(method="PUT", url=url, **kwargs)

async def patch(self, url: str, **kwargs: Unpack[RequestParams]):
async def patch(self, url: str, **kwargs: Unpack[RequestParams]): # type: ignore
return await self.request(method="PATCH", url=url, **kwargs)


Expand Down Expand Up @@ -230,7 +195,7 @@ def request(
headers: an optional map of HTTP headers to send with requests. If `impersonate` is set, this will be ignored.
cookies: an optional map of cookies to send with requests as the `Cookie` header.
timeout: the timeout for the request in seconds. Default is 30.
content: he content to send in the request body as bytes. Default is None.
content: the content to send in the request body as bytes. Default is None.
data: the form data to send in the request body. Default is None.
json: a JSON serializable object to send in the request body. Default is None.
files: a map of file fields to file paths to be sent as multipart/form-data. Default is None.
Expand Down
128 changes: 128 additions & 0 deletions primp/primp.pyi
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
from __future__ import annotations

import sys
from typing import Any, Literal, TypedDict

if sys.version_info <= (3, 11):
from typing_extensions import Unpack
else:
from typing import Unpack

HttpMethod = Literal["GET", "HEAD", "OPTIONS", "DELETE", "POST", "PUT", "PATCH"]
IMPERSONATE = Literal[
"chrome_100", "chrome_101", "chrome_104", "chrome_105", "chrome_106",
"chrome_107", "chrome_108", "chrome_109", "chrome_114", "chrome_116",
"chrome_117", "chrome_118", "chrome_119", "chrome_120", "chrome_123",
"chrome_124", "chrome_126", "chrome_127", "chrome_128", "chrome_129",
"chrome_130", "chrome_131",
"safari_15.3", "safari_15.5", "safari_15.6.1", "safari_16",
"safari_16.5", "safari_17.0", "safari_17.2.1", "safari_17.4.1",
"safari_17.5", "safari_18", "safari_18.2",
"safari_ios_16.5", "safari_ios_17.2", "safari_ios_17.4.1", "safari_ios_18.1.1",
"safari_ipad_18",
"okhttp_3.9", "okhttp_3.11", "okhttp_3.13", "okhttp_3.14", "okhttp_4.9",
"okhttp_4.10", "okhttp_5",
"edge_101", "edge_122", "edge_127", "edge_131",
"firefox_109", "firefox_117", "firefox_128", "firefox_133",
] # fmt: skip
IMPERSONATE_OS = Literal["android", "ios", "linux", "macos", "windows"]

class RequestParams(TypedDict, total=False):
auth: tuple[str, str | None] | None
auth_bearer: str | None
params: dict[str, str] | None
headers: dict[str, str] | None
cookies: dict[str, str] | None
timeout: float | None
content: bytes | None
data: dict[str, Any] | None
json: Any | None
files: dict[str, str] | None

class ClientRequestParams(RequestParams):
impersonate: IMPERSONATE | None
impersonate_os: IMPERSONATE_OS | None
verify: bool | None
ca_cert_file: str | None

class Response:
@property
def content(self) -> bytes: ...
@property
def cookies(self) -> dict[str, str]: ...
@property
def headers(self) -> dict[str, str]: ...
@property
def status_code(self) -> int: ...
@property
def url(self) -> str: ...
@property
def encoding(self) -> str: ...
@property
def text(self) -> str: ...
def json(self) -> Any: ...
@property
def text_markdown(self) -> str: ...
@property
def text_plain(self) -> str: ...
@property
def text_rich(self) -> str: ...

class RClient:
def __init__(
self,
auth: tuple[str, str | None] | None = None,
auth_bearer: str | None = None,
params: dict[str, str] | None = None,
headers: dict[str, str] | None = None,
cookies: dict[str, str] | None = None,
timeout: float | None = None,
cookie_store: bool | None = True,
referer: bool | None = True,
proxy: str | None = None,
impersonate: IMPERSONATE | None = None,
impersonate_os: IMPERSONATE_OS | None = None,
follow_redirects: bool | None = True,
max_redirects: int | None = 20,
verify: bool | None = True,
ca_cert_file: str | None = None,
https_only: bool | None = False,
http2_only: bool | None = False,
): ...
@property
def headers(self) -> dict[str, str]: ...
@headers.setter
def headers(self, headers: dict[str, str]) -> None: ...
@property
def cookies(self) -> dict[str, str]: ...
@cookies.setter
def cookies(self, cookies: dict[str, str]) -> None: ...
@property
def proxy(self) -> str | None: ...
@proxy.setter
def proxy(self, proxy: str) -> None: ...
@property
def impersonate(self) -> str | None: ...
@impersonate.setter
def impersonate(self, impersonate: IMPERSONATE) -> None: ...
@property
def impersonate_os(self) -> str | None: ...
@impersonate_os.setter
def impersonate_os(self, impersonate: IMPERSONATE_OS) -> None: ...
def request(self, method: HttpMethod, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...
def get(self, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...
def head(self, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...
def options(self, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...
def delete(self, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...
def post(self, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...
def put(self, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...
def patch(self, url: str, **kwargs: Unpack[RequestParams]) -> Response: ...

def request(method: HttpMethod, url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...
def get(url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...
def head(url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...
def options(url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...
def delete(url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...
def post(url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...
def put(url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...
def patch(url: str, **kwargs: Unpack[ClientRequestParams]) -> Response: ...