diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 43444ff..865540d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -6,12 +6,12 @@ repos: hooks: - id: taplo - repo: https://github.com/asottile/reorder-python-imports - rev: v3.12.0 + rev: v3.13.0 hooks: - id: reorder-python-imports args: [--py37-plus, --add-import, "from __future__ import annotations"] - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.3.5 + rev: v0.4.9 hooks: - id: ruff types_or: [python, pyi, jupyter] @@ -28,7 +28,7 @@ repos: - id: name-tests-test - id: requirements-txt-fixer - repo: https://github.com/pre-commit/mirrors-mypy - rev: v1.9.0 + rev: v1.10.0 hooks: - id: mypy additional_dependencies: [types-all] diff --git a/pyproject.toml b/pyproject.toml index f88c7ad..2948030 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -153,16 +153,3 @@ docstring-code-line-length = 100 "E501", # line-too-long "INP001", # implicit-namespace-package ] - -[tool.tox] -legacy_tox_ini = """ - [tox] - skip_missing_interpreters = True - envlist = py38, py39, py310, py311, py312 - - [testenv] - deps = pytest - commands = - ; mypy --ignore-missing-imports --scripts-are-modules comma - pytest -W ignore::DeprecationWarning -W ignore::PendingDeprecationWarning {posargs:tests} -""" diff --git a/src/dev_toolbox/cli/ruff_doc.py b/src/dev_toolbox/cli/ruff_doc.py index dabc4be..17cf76d 100644 --- a/src/dev_toolbox/cli/ruff_doc.py +++ b/src/dev_toolbox/cli/ruff_doc.py @@ -1,46 +1,33 @@ from __future__ import annotations -import contextlib import itertools import os -from typing import Generator -from typing import MutableMapping from typing import Sequence -from typing import TYPE_CHECKING from dev_toolbox.cli.html_table import get_column_widths from dev_toolbox.cli.html_table import TablesParser - -if TYPE_CHECKING: - import http.client +from dev_toolbox.http import RequestTemplate +from dev_toolbox.http.great_value import gv_request RUFF_URL = "https://docs.astral.sh/ruff/rules/" _FILE_CACHE = "/tmp/ruff_rules.html" # noqa: S108 -@contextlib.contextmanager -def get_request( - url: str, headers: MutableMapping[str, str] -) -> Generator[http.client.HTTPResponse, None, None]: - import urllib.request - - request = urllib.request.Request(url, headers=headers) # noqa: S310 - - with urllib.request.urlopen(request) as f: # noqa: S310 - yield f +ruff_rules_template = RequestTemplate( + url=RUFF_URL, + headers={ + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:124.0) Gecko/20100101 Firefox/124.0", # noqa: E501 + "Host": "docs.astral.sh", + }, +) def _get_rules_html() -> str: if not os.path.exists(_FILE_CACHE): - headers = { - "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:124.0) Gecko/20100101 Firefox/124.0", # noqa: E501 - "Host": "docs.astral.sh", - } - with get_request( - RUFF_URL, - headers, - ) as response, open(_FILE_CACHE, "wb") as f: + with ruff_rules_template.request(gv_request).response as response, open( + _FILE_CACHE, "wb" + ) as f: f.write(response.read()) with open(_FILE_CACHE) as f: diff --git a/src/dev_toolbox/http/__init__.py b/src/dev_toolbox/http/__init__.py new file mode 100644 index 0000000..d18e662 --- /dev/null +++ b/src/dev_toolbox/http/__init__.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from inspect import isawaitable +from typing import cast +from typing import overload +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from dev_toolbox.http._types import HTTP_METHOD + from dev_toolbox.http._types import _CompleteRequestArgs + from dev_toolbox.http._types import R_co + from dev_toolbox.http._types import _OptionalRequestsArgs + from dev_toolbox.http._types import RequestLike + from dev_toolbox.http._types import RequestLikeAsync + from typing_extensions import Unpack + from typing import Awaitable + from _typeshed import Incomplete + + +class RequestTemplate: + """ + A template for making HTTP requests. + + Args: + ---- + method (HTTP_METHOD): The HTTP method to use for the request. + url (str): The URL to send the request to. + **kwargs: Additional keyword arguments to be passed to the request. + + Attributes: + ---------- + _request_args (RequestLikeArgs): The arguments for the request. + + Methods: + ------- + request: Sends the HTTP request. + json: Sends the HTTP request and returns the response as JSON. + + """ + + __slots__ = ("_request_args",) + _request_args: _CompleteRequestArgs + + def __init__( + self, *, url: str, method: HTTP_METHOD = "GET", **kwargs: Unpack[_OptionalRequestsArgs] + ) -> None: + self._request_args = {"method": method, "url": url, **kwargs} + + @overload + def request(self, http_client: RequestLike[R_co]) -> R_co: ... + + @overload + def request(self, http_client: RequestLikeAsync[R_co]) -> Awaitable[R_co]: ... + + def request( + self, http_client: RequestLike[R_co] | RequestLikeAsync[R_co] + ) -> R_co | Awaitable[R_co]: + """ + Sends the HTTP request. + + Args: + ---- + http_client (RequestLike[R_co] | RequestLikeAsync[R_co]): The HTTP client to use for the request. + + Returns: + ------- + R_co | Awaitable[R_co]: The response from the HTTP request. + + """ # noqa: E501 + return http_client.request(**self._request_args) + + async def __asjon(self, response: Awaitable[R_co]) -> Incomplete: + r = await response + r.raise_for_status() + return r.json() + + @overload + def json(self, http_client: RequestLike[R_co]) -> Incomplete: ... + + @overload + def json(self, http_client: RequestLikeAsync[R_co]) -> Awaitable[Incomplete]: ... + + def json( + self, http_client: RequestLike[R_co] | RequestLikeAsync[R_co] + ) -> Incomplete | Awaitable[Incomplete]: + """ + Sends the HTTP request and returns the response as JSON. + + Args: + ---- + http_client (RequestLike[R_co] | RequestLikeAsync[R_co]): The HTTP client to use for the request. + + Returns: + ------- + Incomplete | Awaitable[Incomplete]: The response from the HTTP request as JSON. + + """ # noqa: E501 + response = self.request(http_client) + if isawaitable(response): + return self.__asjon(response) + response = cast("R_co", response) + response.raise_for_status() + return response.json() diff --git a/src/dev_toolbox/http/_test.py b/src/dev_toolbox/http/_test.py new file mode 100644 index 0000000..b64abaa --- /dev/null +++ b/src/dev_toolbox/http/_test.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from dev_toolbox.http import RequestTemplate + from _typeshed import Incomplete + from typing import Awaitable + + +def test_requests(template: RequestTemplate) -> None: + import requests + + requests_session = requests.Session() + + # y: RequestLike = requests_session + + a = template.request(requests_session) + a = template.json(requests_session) + + r1: requests.Response = template.request(requests_session) + j1: Incomplete = template.json(requests_session) + + # from typing_extensions import reveal_type + # reveal_type(template.request(requests_session)) + # reveal_type(template.json(requests_session)) + + print(a, r1, j1) + + +def test_httpx(template: RequestTemplate) -> None: + import httpx + + # e: RequestLike = httpx_client + # p: RequestLike = httpx_async_client + + httpx_client = httpx.Client() + a = template.request(httpx_client) + a = template.json(httpx_client) + + r2: httpx._models.Response = template.request(httpx_client) + j2: Incomplete = template.json(httpx_client) + + # reveal_type(template.request(httpx_client)) + # reveal_type(template.json(httpx_client)) + print(a, r2, j2) + + +def test_httpx_async(template: RequestTemplate) -> None: + import httpx + + httpx_async_client = httpx.AsyncClient() + a = template.request(httpx_async_client) + a = template.json(httpx_async_client) + + r3: Awaitable[httpx._models.Response] = template.request(httpx_async_client) + j3: Awaitable[Incomplete] = template.json(httpx_async_client) + + # reveal_type(template.request(httpx_async_client)) + # reveal_type(template.json(httpx_async_client)) + + print(a, r3, j3) + + +def test_great_value(template: RequestTemplate) -> None: + from dev_toolbox.http.great_value import GreatValueRequests + from dev_toolbox.http.great_value import GreatValueResponse + + gv_client = GreatValueRequests() + a = template.request(gv_client) + a = template.json(gv_client) + + r3: GreatValueResponse = template.request(gv_client) + j3: Incomplete = template.json(gv_client) + + # reveal_type(template.request(httpx_async_client)) + # reveal_type(template.json(httpx_async_client)) + + print(a, r3, j3) + + +def test_main() -> None: + from dev_toolbox.http import RequestTemplate + + template = RequestTemplate( + url="http://ip.jsontest.com/", + headers={"User-Agent": "Mozilla/5.0"}, + ) + + test_great_value(template) + test_requests(template) + test_httpx(template) + test_httpx_async(template) + + +def test_main2() -> None: + from dev_toolbox.http import RequestTemplate + from dev_toolbox.http.great_value import GreatValueRequests + + client = GreatValueRequests(base_url="https://motionless-hearty-bongo.anvil.app") + + template = RequestTemplate(method="POST", url="/gron", json={"name": "John Doe"}) + response = template.request(client) + print(response.response.read().decode("utf-8")) + + template = RequestTemplate( + url="/get_tables", + params={"url": "https://aws.amazon.com/ec2/instance-types/"}, + ) + response = template.json(client) + print(response) + + +if __name__ == "__main__": + test_main2() diff --git a/src/dev_toolbox/http/_types.py b/src/dev_toolbox/http/_types.py new file mode 100644 index 0000000..9970f81 --- /dev/null +++ b/src/dev_toolbox/http/_types.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from typing import Dict + from typing import List + from typing import Any + from typing import Mapping + from typing import Union + from typing import IO + from typing import Optional + from typing import Tuple + from typing import TypeVar + from typing_extensions import Literal + from typing_extensions import TypedDict + from typing_extensions import Protocol + from typing_extensions import Unpack + + FileContent = Union[IO[bytes], bytes, str] + _FileSpec = Union[ + FileContent, + Tuple[Optional[str], FileContent], + ] + _Params = Union[Dict[str, Any], Tuple[Tuple[str, Any], ...], List[Tuple[str, Any]], None] + + class _OptionalRequestsArgs(TypedDict, total=False): + auth: tuple[str, str] | None + cookies: dict[str, str] | None + data: Mapping[str, Any] | None + files: Mapping[str, _FileSpec] + headers: Mapping[str, Any] | None + json: Any | None + params: _Params + timeout: float | None + + HTTP_METHOD = Literal["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS", "TRACE"] + + class _RequieredRequestsArgs(TypedDict): + method: HTTP_METHOD + url: str + + class _CompleteRequestArgs(_RequieredRequestsArgs, _OptionalRequestsArgs): ... + + class ResponseLike(Protocol): + def json(self) -> Any: ... # noqa: ANN401 + + def raise_for_status(self) -> Any: ... # noqa: ANN401 + + R_co = TypeVar( + "R_co", + covariant=True, # + bound=ResponseLike, # + ) + + class RequestLike(Protocol[R_co]): + def request( + self, method: HTTP_METHOD, url: str, **kwargs: Unpack[_OptionalRequestsArgs] + ) -> R_co: ... + + class RequestLikeAsync(Protocol[R_co]): + async def request( + self, method: HTTP_METHOD, url: str, **kwargs: Unpack[_OptionalRequestsArgs] + ) -> R_co: ... diff --git a/src/dev_toolbox/http/great_value.py b/src/dev_toolbox/http/great_value.py new file mode 100644 index 0000000..dbfe217 --- /dev/null +++ b/src/dev_toolbox/http/great_value.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import json +import urllib.parse +from typing import NamedTuple +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from http.client import HTTPResponse + from _typeshed import Incomplete + from typing_extensions import Unpack + from dev_toolbox.http._types import _Params + from dev_toolbox.http._types import _OptionalRequestsArgs + from dev_toolbox.http._types import HTTP_METHOD + + +class GreatValueResponse(NamedTuple): + response: HTTPResponse + + def json(self) -> Incomplete: + content = self.response.read() + return json.loads(content) + + def raise_for_status(self) -> None: + status_code = self.response.status + http_error_msg = "" + if 400 <= status_code < 500: # noqa: PLR2004 + http_error_msg = ( + f"{status_code} Client Error: {self.response.reason} for url: {self.response.url}" + ) + + elif 500 <= status_code < 600: # noqa: PLR2004 + http_error_msg = ( + f"{status_code} Server Error: {self.response.reason} for url: {self.response.url}" + ) + + if http_error_msg: + raise Exception(http_error_msg) # noqa: TRY002 + + +class GreatValueRequests(NamedTuple): + base_url: str | None = None + unverifiable: bool = True + headers: dict[str, str] | None = None + + def construct_url(self, base_url: str | None, endpoint: str, params: _Params) -> str: + if base_url is not None and not endpoint.startswith("http"): + endpoint = urllib.parse.urljoin(base_url, endpoint) + if params is not None: + encoded = urllib.parse.urlencode(params) + endpoint += "?" + encoded + return endpoint + + def request( + self, method: HTTP_METHOD, url: str, **kwargs: Unpack[_OptionalRequestsArgs] + ) -> GreatValueResponse: + final_url = self.construct_url(self.base_url, url, kwargs.get("params")) + import urllib.request + + headers = { + k.upper(): v + for k, v in (*(self.headers or {}).items(), *(kwargs.get("headers") or {}).items()) + } + + if kwargs.get("data") and kwargs.get("json"): + msg = "Cannot set both 'data' and 'json'" + raise ValueError(msg) + + data = kwargs.get("data") + + json_content = kwargs.get("json") + if json_content is not None: + if "CONTENT-TYPE" not in headers: + headers["CONTENT-TYPE"] = "application/json" + data = json.dumps(json_content).encode("utf-8") # type: ignore[assignment] + + req = urllib.request.Request( # noqa: S310 + url=final_url, + data=None, + headers=headers, + # origin_req_host=None, + unverifiable=self.unverifiable, + method=method, + ) + response: HTTPResponse = urllib.request.urlopen( # noqa: S310 + url=req, + data=data, # type: ignore[arg-type] + timeout=kwargs.get("timeout"), + # cafile=None, + # capath=None, + # cadefault=False, + # context=None, + ) + + return GreatValueResponse(response=response) + + +gv_request = GreatValueRequests()