RequestTemplate: be able to define request without being tied to a li…

…brary (#20)

* RequestTemplate: be able to define request without being tied to a library

.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]

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}
-"""

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:

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()

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()

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: ...