feat(http): switch httpx for niquests in order to stabilize network I/O (#375)

Fixes #365 

Changes proposed in this pull request:

 * Move from httpx to niquests.

Quick history around that, httpx is being a bit stale for a couple of
years now and report of spurious errors like the one in #365 is killing
the global UX of that library. Not only that but httpx is absolutely
unusable in the freethreaded build for example, it crash instantly on
any attempt. Moreover the performance are often pointed out, especially
in async. While Niquests isn't faster than aiohttp right now, it's
better than requests and httpx.

Finally, a touch of hope, we recently were selected into the [GitHub
Secure Open Source
Program](https://github.blog/open-source/maintainers/securing-the-supply-chain-at-scale-starting-with-71-important-open-source-projects/#developer-utilities-and-cli-helpers-%f0%9f%a7%91%f0%9f%92%bb)
(Session 2)! We manage several packages and one very high profile
(charset-normalizer). Our security standards are solid and we currently
are in the way of making every (of our own) package CRA compliant
(charset-normalizer already is!).
 
This PR is a firm start into migrating from httpx, I will take the time
to add comments using GitHub UI so that you can quickly understand why a
change occurred this and there. I tried my best to keep the change
surface as minimal as I could.

Before submitting this, I ensured that the tests were all good.

<img width="643" height="634" alt="image"
src="https://github.com/user-attachments/assets/d17b8a75-cb3e-403c-b1eb-a403d5e2209f"
/>


Regards,

Author: Ousret

docs/conf.py

@@ -70,7 +70,7 @@
     (r"py:class", r"ComputedFieldInfo"),
     (r"py:class", r"FieldInfo"),
     (r"py:class", r"ConfigDict"),
-    (r"py:.*", r"httpx.*"),
+    (r"py:.*", r"niquests.*"),
 ]
 
 autodoc_member_order = "bysource"

examples/as_app/talk_bot/lib/main.py

@@ -4,7 +4,7 @@
 from contextlib import asynccontextmanager
 from typing import Annotated
 
-import httpx
+import niquests
 from fastapi import BackgroundTasks, Depends, FastAPI, Response
 
 from nc_py_api import NextcloudApp, talk_bot
@@ -32,7 +32,7 @@ def convert_currency(amount, from_currency, to_currency):
     base_url = "https://api.exchangerate-api.com/v4/latest/"
 
     # Fetch latest exchange rates
-    response = httpx.get(base_url + from_currency, timeout=60)
+    response = niquests.get(base_url + from_currency, timeout=60)
     data = response.json()
 
     if "rates" in data:

nc_py_api/_exceptions.py

@@ -1,6 +1,6 @@
 """Exceptions for the Nextcloud API."""
 
-from httpx import Response, codes
+from niquests import HTTPError, Response
 
 
 class NextcloudException(Exception):
@@ -60,6 +60,8 @@ def check_error(response: Response, info: str = ""):
         else:
             phrase = "Unknown error"
         raise NextcloudException(status_code, reason=phrase, info=info)
-    if not codes.is_error(status_code):
-        return
-    raise NextcloudException(status_code, reason=codes(status_code).phrase, info=info)
+
+    try:
+        response.raise_for_status()
+    except HTTPError as e:
+        raise NextcloudException(status_code, reason=response.reason, info=info) from e

nc_py_api/_session.py

@@ -10,9 +10,11 @@
 from enum import IntEnum
 from json import loads
 from os import environ
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
-from httpx import AsyncClient, Client, Headers, Limits, ReadTimeout, Request, Response
-from httpx import __version__ as httpx_version
+from niquests import AsyncSession, ReadTimeout, Request, Response, Session
+from niquests import __version__ as niquests_version
+from niquests.structures import CaseInsensitiveDict
 from starlette.requests import HTTPConnection
 
 from . import options
@@ -49,6 +51,13 @@ class ServerVersion(typing.TypedDict):
     """Indicates if the subscription has extended support"""
 
 
+@dataclass
+class Limits:
+    max_keepalive_connections: int | None = 20
+    max_connections: int | None = 100
+    keepalive_expiry: int | float | None = 5
+
+
 @dataclass
 class RuntimeOptions:
     xdebug_session: str
@@ -134,11 +143,11 @@ def __init__(self, **kwargs):
 
 
 class NcSessionBase(ABC):
-    adapter: AsyncClient | Client
-    adapter_dav: AsyncClient | Client
+    adapter: AsyncSession | Session
+    adapter_dav: AsyncSession | Session
     cfg: BasicConfig
     custom_headers: dict
-    response_headers: Headers
+    response_headers: CaseInsensitiveDict
     _user: str
     _capabilities: dict
 
@@ -150,7 +159,7 @@ def __init__(self, **kwargs):
         self.limits = Limits(max_keepalive_connections=20, max_connections=20, keepalive_expiry=60.0)
         self.init_adapter()
         self.init_adapter_dav()
-        self.response_headers = Headers()
+        self.response_headers = CaseInsensitiveDict()
         self._ocs_regexp = re.compile(r"/ocs/v[12]\.php/|/apps/groupfolders/")
 
     def init_adapter(self, restart=False) -> None:
@@ -172,7 +181,7 @@ def init_adapter_dav(self, restart=False) -> None:
                 self.adapter_dav.cookies.set("XDEBUG_SESSION", options.XDEBUG_SESSION)
 
     @abstractmethod
-    def _create_adapter(self, dav: bool = False) -> AsyncClient | Client:
+    def _create_adapter(self, dav: bool = False) -> AsyncSession | Session:
         pass  # pragma: no cover
 
     @property
@@ -187,8 +196,8 @@ def ae_url_v2(self) -> str:
 
 
 class NcSessionBasic(NcSessionBase, ABC):
-    adapter: Client
-    adapter_dav: Client
+    adapter: Session
+    adapter_dav: Session
 
     def ocs(
         self,
@@ -206,9 +215,7 @@ def ocs(
         info = f"request: {method} {path}"
         nested_req = kwargs.pop("nested_req", False)
         try:
-            response = self.adapter.request(
-                method, path, content=content, json=json, params=params, files=files, **kwargs
-            )
+            response = self.adapter.request(method, path, data=content, json=json, params=params, files=files, **kwargs)
         except ReadTimeout:
             raise NextcloudException(408, info=info) from None
 
@@ -281,18 +288,18 @@ def _get_adapter_kwargs(self, dav: bool) -> dict[str, typing.Any]:
             return {
                 "base_url": self.cfg.dav_endpoint,
                 "timeout": self.cfg.options.timeout_dav,
-                "event_hooks": {"request": [], "response": [self._response_event]},
+                "event_hooks": {"pre_request": [], "response": [self._response_event]},
             }
         return {
             "base_url": self.cfg.endpoint,
             "timeout": self.cfg.options.timeout,
-            "event_hooks": {"request": [self._request_event_ocs], "response": [self._response_event]},
+            "event_hooks": {"pre_request": [self._request_event_ocs], "response": [self._response_event]},
         }
 
     def _request_event_ocs(self, request: Request) -> None:
         str_url = str(request.url)
         if re.search(self._ocs_regexp, str_url) is not None:  # this is OCS call
-            request.url = request.url.copy_merge_params({"format": "json"})
+            request.url = patch_param(request.url, "format", "json")
             request.headers["Accept"] = "application/json"
 
     def _response_event(self, response: Response) -> None:
@@ -305,15 +312,15 @@ def _response_event(self, response: Response) -> None:
 
     def download2fp(self, url_path: str, fp, dav: bool, params=None, **kwargs):
         adapter = self.adapter_dav if dav else self.adapter
-        with adapter.stream("GET", url_path, params=params, headers=kwargs.get("headers")) as response:
+        with adapter.get(url_path, params=params, headers=kwargs.get("headers"), stream=True) as response:
             check_error(response)
-            for data_chunk in response.iter_bytes(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
+            for data_chunk in response.iter_raw(chunk_size=kwargs.get("chunk_size", -1)):
                 fp.write(data_chunk)
 
 
 class AsyncNcSessionBasic(NcSessionBase, ABC):
-    adapter: AsyncClient
-    adapter_dav: AsyncClient
+    adapter: AsyncSession
+    adapter_dav: AsyncSession
 
     async def ocs(
         self,
@@ -332,7 +339,7 @@ async def ocs(
         nested_req = kwargs.pop("nested_req", False)
         try:
             response = await self.adapter.request(
-                method, path, content=content, json=json, params=params, files=files, **kwargs
+                method, path, data=content, json=json, params=params, files=files, **kwargs
             )
         except ReadTimeout:
             raise NextcloudException(408, info=info) from None
@@ -350,7 +357,7 @@ async def ocs(
                 and ocs_meta["statuscode"] == 403
                 and str(ocs_meta["message"]).lower().find("password confirmation is required") != -1
             ):
-                await self.adapter.aclose()
+                await self.adapter.close()
                 self.init_adapter(restart=True)
                 return await self.ocs(
                     method, path, **kwargs, content=content, json=json, params=params, nested_req=True
@@ -408,18 +415,18 @@ def _get_adapter_kwargs(self, dav: bool) -> dict[str, typing.Any]:
             return {
                 "base_url": self.cfg.dav_endpoint,
                 "timeout": self.cfg.options.timeout_dav,
-                "event_hooks": {"request": [], "response": [self._response_event]},
+                "event_hooks": {"pre_request": [], "response": [self._response_event]},
             }
         return {
             "base_url": self.cfg.endpoint,
             "timeout": self.cfg.options.timeout,
-            "event_hooks": {"request": [self._request_event_ocs], "response": [self._response_event]},
+            "event_hooks": {"pre_request": [self._request_event_ocs], "response": [self._response_event]},
         }
 
     async def _request_event_ocs(self, request: Request) -> None:
         str_url = str(request.url)
         if re.search(self._ocs_regexp, str_url) is not None:  # this is OCS call
-            request.url = request.url.copy_merge_params({"format": "json"})
+            request.url = patch_param(request.url, "format", "json")
             request.headers["Accept"] = "application/json"
 
     async def _response_event(self, response: Response) -> None:
@@ -432,10 +439,12 @@ async def _response_event(self, response: Response) -> None:
 
     async def download2fp(self, url_path: str, fp, dav: bool, params=None, **kwargs):
         adapter = self.adapter_dav if dav else self.adapter
-        async with adapter.stream("GET", url_path, params=params, headers=kwargs.get("headers")) as response:
-            check_error(response)
-            async for data_chunk in response.aiter_bytes(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
-                fp.write(data_chunk)
+        response = await adapter.get(url_path, params=params, headers=kwargs.get("headers"), stream=True)
+
+        check_error(response)
+
+        async for data_chunk in await response.iter_raw(chunk_size=kwargs.get("chunk_size", -1)):
+            fp.write(data_chunk)
 
 
 class NcSession(NcSessionBasic):
@@ -445,15 +454,20 @@ def __init__(self, **kwargs):
         self.cfg = Config(**kwargs)
         super().__init__()
 
-    def _create_adapter(self, dav: bool = False) -> AsyncClient | Client:
-        return Client(
-            follow_redirects=True,
-            limits=self.limits,
-            verify=self.cfg.options.nc_cert,
-            **self._get_adapter_kwargs(dav),
-            auth=self.cfg.auth,
+    def _create_adapter(self, dav: bool = False) -> AsyncSession | Session:
+        session_kwargs = self._get_adapter_kwargs(dav)
+        hooks = session_kwargs.pop("event_hooks")
+
+        session = Session(
+            keepalive_delay=self.limits.keepalive_expiry, pool_maxsize=self.limits.max_connections, **session_kwargs
         )
 
+        session.auth = self.cfg.auth
+        session.verify = self.cfg.options.nc_cert
+        session.hooks.update(hooks)
+
+        return session
+
 
 class AsyncNcSession(AsyncNcSessionBasic):
     cfg: Config
@@ -462,21 +476,28 @@ def __init__(self, **kwargs):
         self.cfg = Config(**kwargs)
         super().__init__()
 
-    def _create_adapter(self, dav: bool = False) -> AsyncClient | Client:
-        return AsyncClient(
-            follow_redirects=True,
-            limits=self.limits,
-            verify=self.cfg.options.nc_cert,
-            **self._get_adapter_kwargs(dav),
-            auth=self.cfg.auth,
+    def _create_adapter(self, dav: bool = False) -> AsyncSession | Session:
+        session_kwargs = self._get_adapter_kwargs(dav)
+        hooks = session_kwargs.pop("event_hooks")
+
+        session = AsyncSession(
+            keepalive_delay=self.limits.keepalive_expiry,
+            pool_maxsize=self.limits.max_connections,
+            **session_kwargs,
         )
 
+        session.verify = self.cfg.options.nc_cert
+        session.auth = self.cfg.auth
+        session.hooks.update(hooks)
+
+        return session
+
 
 class NcSessionAppBasic(ABC):
     cfg: AppConfig
     _user: str
-    adapter: AsyncClient | Client
-    adapter_dav: AsyncClient | Client
+    adapter: AsyncSession | Session
+    adapter_dav: AsyncSession | Session
 
     def __init__(self, **kwargs):
         self.cfg = AppConfig(**kwargs)
@@ -505,22 +526,29 @@ def sign_check(self, request: HTTPConnection) -> str:
 class NcSessionApp(NcSessionAppBasic, NcSessionBasic):
     cfg: AppConfig
 
-    def _create_adapter(self, dav: bool = False) -> AsyncClient | Client:
-        r = self._get_adapter_kwargs(dav)
-        r["event_hooks"]["request"].append(self._add_auth)
-        return Client(
-            follow_redirects=True,
-            limits=self.limits,
-            verify=self.cfg.options.nc_cert,
-            **r,
-            headers={
-                "AA-VERSION": self.cfg.aa_version,
-                "EX-APP-ID": self.cfg.app_name,
-                "EX-APP-VERSION": self.cfg.app_version,
-                "user-agent": f"ExApp/{self.cfg.app_name}/{self.cfg.app_version} (httpx/{httpx_version})",
-            },
+    def _create_adapter(self, dav: bool = False) -> AsyncSession | Session:
+        session_kwargs = self._get_adapter_kwargs(dav)
+        session_kwargs["event_hooks"]["pre_request"].append(self._add_auth)
+
+        hooks = session_kwargs.pop("event_hooks")
+
+        session = Session(
+            keepalive_delay=self.limits.keepalive_expiry,
+            pool_maxsize=self.limits.max_connections,
+            **session_kwargs,
         )
 
+        session.verify = self.cfg.options.nc_cert
+        session.headers = {
+            "AA-VERSION": self.cfg.aa_version,
+            "EX-APP-ID": self.cfg.app_name,
+            "EX-APP-VERSION": self.cfg.app_version,
+            "user-agent": f"ExApp/{self.cfg.app_name}/{self.cfg.app_version} (niquests/{niquests_version})",
+        }
+        session.hooks.update(hooks)
+
+        return session
+
     def _add_auth(self, request: Request):
         request.headers.update(
             {"AUTHORIZATION-APP-API": b64encode(f"{self._user}:{self.cfg.app_secret}".encode("UTF=8"))}
@@ -530,23 +558,39 @@ def _add_auth(self, request: Request):
 class AsyncNcSessionApp(NcSessionAppBasic, AsyncNcSessionBasic):
     cfg: AppConfig
 
-    def _create_adapter(self, dav: bool = False) -> AsyncClient | Client:
-        r = self._get_adapter_kwargs(dav)
-        r["event_hooks"]["request"].append(self._add_auth)
-        return AsyncClient(
-            follow_redirects=True,
-            limits=self.limits,
-            verify=self.cfg.options.nc_cert,
-            **r,
-            headers={
-                "AA-VERSION": self.cfg.aa_version,
-                "EX-APP-ID": self.cfg.app_name,
-                "EX-APP-VERSION": self.cfg.app_version,
-                "User-Agent": f"ExApp/{self.cfg.app_name}/{self.cfg.app_version} (httpx/{httpx_version})",
-            },
+    def _create_adapter(self, dav: bool = False) -> AsyncSession | Session:
+        session_kwargs = self._get_adapter_kwargs(dav)
+        session_kwargs["event_hooks"]["pre_request"].append(self._add_auth)
+
+        hooks = session_kwargs.pop("event_hooks")
+
+        session = AsyncSession(
+            keepalive_delay=self.limits.keepalive_expiry,
+            pool_maxsize=self.limits.max_connections,
+            **session_kwargs,
         )
+        session.verify = self.cfg.options.nc_cert
+        session.headers = {
+            "AA-VERSION": self.cfg.aa_version,
+            "EX-APP-ID": self.cfg.app_name,
+            "EX-APP-VERSION": self.cfg.app_version,
+            "User-Agent": f"ExApp/{self.cfg.app_name}/{self.cfg.app_version} (niquests/{niquests_version})",
+        }
+        session.hooks.update(hooks)
+
+        return session
 
     async def _add_auth(self, request: Request):
         request.headers.update(
             {"AUTHORIZATION-APP-API": b64encode(f"{self._user}:{self.cfg.app_secret}".encode("UTF=8"))}
         )
+
+
+def patch_param(url: str, key: str, value: str) -> str:
+    parts = urlsplit(url)
+    query = dict(parse_qsl(parts.query, keep_blank_values=True))
+    query[key] = value
+
+    new_query = urlencode(query, doseq=True)
+
+    return urlunsplit((parts.scheme, parts.netloc, parts.path, new_query, parts.fragment))