Repository: RafaelWO/unparallel (language: python)

Issue:
Support configuring HTTPX limits and timeouts
## 🚀 Feature Request

Currently, one cannot configure limit or timeout settings for the HTTPX client used in the background. It would be useful to have options in the `up()` method to set those.

## 🔈 Motivation
If a user needs specific connection limits or different timeout settings, this user would be sad as this is currently not supported.

## 🛰 Alternatives

n/a

## 📎 Additional context

See also HTTPX [timeout](https://www.python-httpx.org/advanced/#timeout-configuration) and [pool limit](https://www.python-httpx.org/advanced/#pool-limit-configuration) config.



Discussion / hints:

Fix patch:
diff --git a/Makefile b/Makefile
index 7327624..a8303ab 100644
--- a/Makefile
+++ b/Makefile
@@ -50,7 +50,7 @@ check-codestyle:
 
 .PHONY: check-typing
 check-typing:
-	poetry run mypy --config-file pyproject.toml ./
+	poetry run mypy --config-file pyproject.toml unparallel
 
 .PHONY: check-safety
 check-safety:
diff --git a/README.md b/README.md
index f579a76..17ac954 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@
 
 <div align="center" markdown="1">
 
-Create Python async web requests in no time with **unparallel**!
+Create async HTTP requests with Python in no time.
 
 [![Build status](https://github.com/RafaelWO/unparallel/actions/workflows/test.yml/badge.svg?branch=main&event=push)](https://github.com/RafaelWO/unparallel/actions?query=workflow%3Atest)
 ![Coverage Report](https://raw.githubusercontent.com/RafaelWO/unparallel/main/assets/images/coverage.svg)
@@ -19,6 +19,10 @@ Create Python async web requests in no time with **unparallel**!
 
 </div>
 
+With Unparallel you can easily create thousands of web requests in an efficient way leveraging Python's async capabilities.
+
+Unparallel is built on top of [HTTPX](https://github.com/encode/httpx/) and aims to support its rich set of features.
+
 ## Installation
 
 ```bash
diff --git a/docs/reference.md b/docs/reference.md
index ee944b5..582273c 100644
--- a/docs/reference.md
+++ b/docs/reference.md
@@ -1,12 +1,1 @@
 ::: unparallel
-    options:
-      show_root_heading: true
-      
-
-::: unparallel.unparallel
-    options:
-      show_root_heading: true
-      members:
-        - sort_by_idx
-        - single_request
-        - request_urls
diff --git a/mkdocs.yml b/mkdocs.yml
index 2f6b951..bb7e12d 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -35,7 +35,11 @@ plugins:
   - mkdocstrings:
       handlers:
         python:
-          members_order: source
+          options:
+            show_root_heading: true
+            members_order: source
+            show_root_members_full_path: true
+            separate_signature: true
 markdown_extensions:
   - md_in_html
   - pymdownx.highlight:
diff --git a/pyproject.toml b/pyproject.toml
index c6f6b66..1a64797 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -86,8 +86,6 @@ exclude = '''
 py_version = 38
 line_length = 88
 
-known_typing = ["typing", "types", "typing_extensions", "mypy", "mypy_extensions"]
-sections = ["FUTURE", "TYPING", "STDLIB", "THIRDPARTY", "FIRSTPARTY", "LOCALFOLDER"]
 include_trailing_comma = true
 profile = "black"
 multi_line_output = 3
diff --git a/unparallel/__init__.py b/unparallel/__init__.py
index 94489b6..7682965 100644
--- a/unparallel/__init__.py
+++ b/unparallel/__init__.py
@@ -1,5 +1,3 @@
-"""Create async web requests in no time."""
-
 from importlib import metadata as importlib_metadata
 
 from .unparallel import RequestError, up
diff --git a/unparallel/unparallel.py b/unparallel/unparallel.py
index 8844568..52030a2 100644
--- a/unparallel/unparallel.py
+++ b/unparallel/unparallel.py
@@ -1,14 +1,16 @@
-from typing import Any, Dict, List, Optional, Tuple, Union
-
 import asyncio
 import logging
 from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple, Union
 
-from httpx import AsyncClient, Limits, TimeoutException
+import httpx
 from tqdm.asyncio import tqdm as tqdm_async
 
 logger = logging.getLogger(__name__)
-VALID_HTTP_METHODS = ("get", "options", "head", "post", "put", "patch", "delete")
+VALID_HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "HEAD", "PATCH", "OPTIONS")
+
+DEFAULT_TIMEOUT = httpx.Timeout(timeout=10)
+DEFAULT_LIMITS = httpx.Limits(max_connections=100, max_keepalive_connections=20)
 
 
 @dataclass
@@ -39,7 +41,7 @@ def sort_by_idx(results: List[Tuple[int, Any]]) -> List[Any]:
 
 async def single_request(
     idx: int,
-    client: AsyncClient,
+    client: httpx.AsyncClient,
     path: str,
     method: str,
     json: Optional[Any] = None,
@@ -48,10 +50,10 @@ async def single_request(
     """Do a single web request for the given path, HTTP method, and playload.
 
     Args:
-        idx (int): The index of the task (required for sorting afterwards)
-        client (AsyncClient): The httpx client
-        path (str): The path after the base URI
-        method (str): The HTTP method
+        idx (int): The index of the task (required for sorting afterwards).
+        client (AsyncClient): The httpx client.
+        path (str): The path after the base URI.
+        method (str): The HTTP method.
         json (Optional[Any], optional): The JSON payload. Defaults to None.
         max_retries_on_timeout (int): The maximum number retries if the requests fails
             due to a timeout (``httpx.TimeoutException``). Defauls to 3.
@@ -61,17 +63,13 @@ async def single_request(
     """
     trial = 0
     exception: Optional[Exception] = None
-    method = method.lower()
     for trial in range(1, max_retries_on_timeout + 1):
         try:
-            kwargs = {}
-            if method in ["post", "put", "patch"]:
-                kwargs["json"] = json
-            response = await getattr(client, method.lower())(path, **kwargs)
+            response = await client.request(method, path, json=json)
             response.raise_for_status()
             json_data = response.json()
             return idx, json_data
-        except TimeoutException as timeout_ex:
+        except httpx.TimeoutException as timeout_ex:
             exception = timeout_ex
             await asyncio.sleep(1)
         except Exception as ex:  # pylint: disable=broad-except
@@ -97,12 +95,13 @@ async def single_request(
 async def request_urls(
     base_url: str,
     paths: Union[str, List[str]],
-    method: str = "get",
+    method: str,
     headers: Optional[Dict[str, Any]] = None,
     payloads: Optional[Any] = None,
     flatten_result: bool = False,
-    connection_limit: int = 100,
     max_retries_on_timeout: int = 3,
+    limits: httpx.Limits = DEFAULT_LIMITS,
+    timeouts: httpx.Timeout = DEFAULT_TIMEOUT,
     progress: bool = True,
 ) -> List[Any]:
     """
@@ -119,9 +118,10 @@ async def request_urls(
             Used together with paths.
         flatten_result: If True and the response per request is a list, flatten that
             list of lists. This is useful when using paging.
-        connection_limit: The total number of simultaneous TCP connections
         max_retries_on_timeout (int): The maximum number retries if the requests fails
             due to a timeout (``httpx.TimeoutException``). Defauls to 3.
+        limits (httpx.Limits): The limits configuration for ``httpx``.
+        timeouts (httpx.Timeout): The timeout configuration for ``httpx``.
         progress: If set to True, progress bar is shown
 
     Returns:
@@ -131,8 +131,13 @@ async def request_urls(
     tasks = []
     results = []
 
-    limits = Limits(max_connections=connection_limit)
-    async with AsyncClient(base_url=base_url, headers=headers, limits=limits) as client:
+    logging.debug(
+        f"Issuing {len(paths)} {method.upper()} request(s) to base URL '{base_url}' "
+        f"with {limits.max_connections} max connections..."
+    )
+    async with httpx.AsyncClient(
+        base_url=base_url, headers=headers, timeout=timeouts, limits=limits
+    ) as client:
         for i, path in enumerate(paths):
             task = asyncio.create_task(
                 single_request(
@@ -141,6 +146,7 @@ async def request_urls(
                     client=client,
                     method=method,
                     json=payloads[i] if payloads else None,
+                    max_retries_on_timeout=max_retries_on_timeout,
                 )
             )
             tasks.append(task)
@@ -160,12 +166,15 @@ async def request_urls(
 async def up(
     base_url: str,
     paths: Union[str, List[str]],
-    method: str = "get",
+    method: str = "GET",
     headers: Optional[Dict[str, Any]] = None,
     payloads: Optional[Any] = None,
     flatten_result: bool = False,
-    connection_limit: int = 100,
+    max_connections: Optional[int] = 100,
+    timeout: Optional[int] = 10,
     max_retries_on_timeout: int = 3,
+    limits: Optional[httpx.Limits] = None,
+    timeouts: Optional[httpx.Timeout] = None,
     progress: bool = True,
 ) -> List[Any]:
     """Creates async web requests to a URL at the specified path(s) via ``asyncio``
@@ -176,8 +185,8 @@ async def up(
         paths (Union[str, List[str]]): One path or a list of paths, e.g. /foobar/.
             If one path but multiple payloads are supplied, that path is used for all
             requests.
-        method (str): HTTP method to use, e.g. get, post, etc.
-            Defaults to "get".
+        method (str): HTTP method to use - one of ``GET``, ``OPTIONS``, ``HEAD``,
+            ``POST``, ``PUT``, ``PATCH``, or ``DELETE``. Defaults to ``GET``.
         headers (Optional[Dict[str, Any]], optional): A dictionary of headers to use.
             Defaults to None.
         payloads (Optional[Any], optional): A list of JSON payloads (dictionaries) e.g.
@@ -185,10 +194,16 @@ async def up(
         flatten_result (bool): If True and the response per request is a list,
             flatten that list of lists. This is useful when using paging.
             Defaults to False.
-        connection_limit (int): The total number of simultaneous TCP
-            connections. Defaults to 100.
+        max_connections (int): The total number of simultaneous TCP
+            connections. Defaults to 100. This is passed into ``httpx.Limits``.
+        timeout (int): The timeout for requests in seconds. Defaults to 10.
+            This is passed into ``httpx.Timeout``.
         max_retries_on_timeout (int): The maximum number retries if the requests fails
             due to a timeout (``httpx.TimeoutException``). Defauls to 3.
+        limits (Optional[httpx.Limits]): The limits configuration for ``httpx``.
+            If specified, this overrides the ``max_connections`` parameter.
+        timeouts (Optional[httpx.Timeout]): The timeout configuration for ``httpx``.
+            If specified, this overrides the ``timeout`` parameter.
         progress (bool): If set to True, progress bar is shown.
             Defaults to True.
 
@@ -203,7 +218,7 @@ async def up(
         input (paths/payloads).
     """
     # Check if method it valid
-    if method not in VALID_HTTP_METHODS:
+    if method.upper() not in VALID_HTTP_METHODS:
         raise ValueError(
             f"The method '{method}' is not a supported HTTP method. "
             f"Supported methods: {VALID_HTTP_METHODS}"
@@ -222,10 +237,15 @@ async def up(
                 f"{len(paths)} != {len(payloads)}"
             )
 
-    logging.debug(
-        f"Issuing {len(paths)} {method.upper()} request(s) to base URL '{base_url}' "
-        f"with {connection_limit} max connections..."
-    )
+    if timeouts is None:
+        timeouts = httpx.Timeout(timeout)
+    if limits is None:
+        if max_connections != DEFAULT_LIMITS.max_connections:
+            limits = httpx.Limits(max_connections=max_connections)
+            limits.max_keepalive_connections = DEFAULT_LIMITS.max_keepalive_connections
+        else:
+            limits = DEFAULT_LIMITS
+
     return await request_urls(
         base_url=base_url,
         paths=paths,
@@ -233,7 +253,8 @@ async def up(
         headers=headers,
         payloads=payloads,
         flatten_result=flatten_result,
-        connection_limit=connection_limit,
         max_retries_on_timeout=max_retries_on_timeout,
         progress=progress,
+        limits=limits,
+        timeouts=timeouts,
     )


Test patch:
diff --git a/tests/test_unparallel.py b/tests/test_unparallel.py
index 9dc1c04..94cd750 100644
--- a/tests/test_unparallel.py
+++ b/tests/test_unparallel.py
@@ -2,11 +2,14 @@
 import time
 from unittest import mock
 
+import httpx
 import pytest
 from httpx import AsyncClient, Response, TimeoutException
 
 from unparallel import up
 from unparallel.unparallel import (
+    DEFAULT_LIMITS,
+    DEFAULT_TIMEOUT,
     RequestError,
     request_urls,
     single_request,
@@ -42,7 +45,7 @@ async def test_single_request_fail(status, respx_mock):
     url = "http://test.com/foo"
     respx_mock.get(url).mock(return_value=Response(status))
     session = AsyncClient()
-    result = await single_request(1, session, path=url, method="get")
+    result = await single_request(1, session, path=url, method="GET")
     assert isinstance(result[1], RequestError)
     await session.aclose()
 
@@ -55,7 +58,7 @@ async def test_single_request_timeout(respx_mock):
     start_time = time.time()
     retries = 2
     result = await single_request(
-        1, session, path=url, method="get", max_retries_on_timeout=retries
+        1, session, path=url, method="GET", max_retries_on_timeout=retries
     )
     assert isinstance(result[1], RequestError)
     assert time.time() - start_time > retries
@@ -66,7 +69,7 @@ async def test_single_request_timeout(respx_mock):
 async def test_full_run_via_httpbin():
     url = "https://httpbin.org"
     paths = [f"/get?i={i}" for i in range(20)]
-    results = await up(url, paths, "get", connection_limit=10)
+    results = await up(url, paths, "get", max_connections=10)
     assert len(results) == 20
     assert all(res["args"]["i"] == str(i) for i, res in enumerate(results))
 
@@ -82,7 +85,7 @@ async def test_up_get(caplog, respx_mock):
             return_value=Response(200, json={key: int(val)})
         )
 
-    results = await up(base_url, paths, "get")
+    results = await up(base_url, paths, "GET")
 
     my_log = next(rec for rec in caplog.records if rec.module == "unparallel")
     assert "Issuing 3 GET request(s)" in my_log.message
@@ -122,7 +125,7 @@ async def test_request_urls_flat(patched_fetch):
 async def test_up_misaligned_paths_and_payloads():
     with pytest.raises(ValueError):
         await up(
-            "http://test.com", paths=["/a", "/b"], method="post", payloads=[1, 2, 3]
+            "http://test.com", paths=["/a", "/b"], method="POST", payloads=[1, 2, 3]
         )
 
 
@@ -130,3 +133,39 @@ async def test_up_misaligned_paths_and_payloads():
 async def test_up_wrong_method():
     with pytest.raises(ValueError):
         await up("http://test.com", paths="/a", method="foobar")
+
+
+@pytest.mark.parametrize(
+    "up_kwargs, expected_timeouts, expected_limits",
+    [
+        ({}, DEFAULT_TIMEOUT, DEFAULT_LIMITS),
+        (
+            {"timeout": 3, "max_connections": 10},
+            httpx.Timeout(3),
+            httpx.Limits(
+                max_connections=10,
+                max_keepalive_connections=DEFAULT_LIMITS.max_keepalive_connections,
+            ),
+        ),
+        (
+            {
+                "timeout": 100,
+                "timeouts": httpx.Timeout(2, connect=5),
+                "limits": httpx.Limits(max_connections=300),
+            },
+            httpx.Timeout(2, connect=5),
+            httpx.Limits(max_connections=300),
+        ),
+    ],
+)
+@mock.patch("unparallel.unparallel.request_urls")
+@pytest.mark.asyncio
+async def test_config(request_urls_mock, up_kwargs, expected_timeouts, expected_limits):
+    def get_kwargs(*args, **kwargs):
+        return kwargs
+
+    request_urls_mock.side_effect = get_kwargs
+
+    options = await up("foobar", "/bar", **up_kwargs)
+    assert options["timeouts"] == expected_timeouts
+    assert options["limits"] == expected_limits