Repository: RafaelWO/unparallel (language: python)

Issue:
Make response processing configurable
## 🚀 Feature Request

Currently, Unparallel will always return the output of `response.json()` for every request. However, one would like just to get the status code or the HTML/text result.

Add a parameter to the `up(...)` to control the returned response. We could use the [`httpx.Response`](https://www.python-httpx.org/api/#response) as the type for a parameter `response`. E.g. the default would be `response=httpx.Response.json`. The parameter type should be a callable and should also allow lambdas for custom control of the results.

## 🔈 Motivation

For example, [this user](https://stackoverflow.com/questions/63688616/asynchronous-status-check-on-multiple-http-requests) wants to return True/False based on the status code. Unparallel should support this use case.

Moreover, one cannot use the supported HTTP method `HEAD` now because it has no JSON body. Having this feature would enable it.

## 🛰 Alternatives

n/a

## 📎 Additional context

n/a



Discussion / hints:

Fix patch:
diff --git a/docs/examples.md b/docs/examples.md
index 56f4b09..a202f4a 100644
--- a/docs/examples.md
+++ b/docs/examples.md
@@ -5,7 +5,7 @@ This page contains some more advanced and realistic examples of using Unparallel
 --8<-- "docs/examples/wordpress.py"
 ```
 
-This should print something like:
+If you run the example, it should print something like the following:
 ```
 Website 'https://techcrunch.com/wp-json/wp/v2' has 12202 pages (page size = 20)
 Making async requests: 100%|███████████| 1000/1000 [00:13<00:00, 74.30it/s]
@@ -40,3 +40,18 @@ Making async requests: 100%|███████████| 1000/1000 [00:13<
                         'over payment dues'},
   'author': 133574269}]
 ```
+
+## Fetch the content of multiple websites
+```python
+--8<-- "docs/examples/multiple_websites.py"
+```
+
+If you run the example, it should print something like the following:
+```
+Making async requests: 100%|███████████| 43/43 [00:03<00:00, 11.60it/s]
+https://www.google.com/ '<!doctype html><html itemscope="" itemtype="http://schema.org/WebPage" lang="de-AT"><head><meta cont'
+https://www.youtube.com/ '<!DOCTYPE html><html style="font-size: 10px;font-family: Roboto, Arial, sans-serif;" lang="de-DE" da'
+https://www.facebook.com/ '<!DOCTYPE html>\n<html lang="de" id="facebook" class="no_js">\n<head><meta charset="utf-8" /><meta nam'
+https://www.wikipedia.org/ '<!DOCTYPE html>\n<html lang="en" class="no-js">\n<head>\n<meta charset="utf-8">\n<title>Wikipedia</title'
+https://www.instagram.com/ '<!DOCTYPE html><html class="_9dls" lang="en" dir="ltr"><head><link data-default-icon="https://static'
+```
diff --git a/docs/examples/multiple_websites.py b/docs/examples/multiple_websites.py
new file mode 100644
index 0000000..6a3e9fe
--- /dev/null
+++ b/docs/examples/multiple_websites.py
@@ -0,0 +1,72 @@
+"""
+This script is based on a use case at StackOverflow and fetches the content of various 
+webpages.
+
+See Also:
+    https://www.stackoverflow.com/a/57129241
+"""
+import asyncio
+
+from unparallel import up
+from unparallel.unparallel import RequestError
+
+# Extracted from https://en.wikipedia.org/wiki/List_of_most-visited_websites
+websites = """https://www.google.com/
+https://www.youtube.com/
+https://www.facebook.com/
+https://www.wikipedia.org/
+https://www.instagram.com/
+https://www.reddit.com/
+https://www.amazon.com/
+https://www.duckduckgo.com/
+https://www.yahoo.com/
+https://www.tiktok.com/
+https://www.bing.com/
+https://www.yahoo.co.jp/
+https://www.weather.com/
+https://www.whatsapp.com/
+https://www.yandex.ru/
+https://www.openai.com/
+https://www.live.com/
+https://www.microsoft.com/
+https://www.linkedin.com/
+https://www.quora.com/
+https://www.twitch.tv/
+https://www.naver.com/
+https://www.netflix.com/
+https://www.office.com/
+https://www.vk.com/
+https://www.globo.com/
+https://www.Aliexpress.com/
+https://www.cnn.com/
+https://www.zoom.us/
+https://www.imdb.com/
+https://www.x.com/
+https://www.nytimes.com/
+https://www.espn.com/
+https://www.amazon.co.jp/
+https://www.pinterest.com/
+https://www.uol.com.br/
+https://www.ebay.com/
+https://www.marca.com/
+https://www.canva.com/
+https://www.spotify.com/
+https://www.bbc.com/
+https://www.paypal.com/
+https://www.apple.com/"""
+
+
+async def main():
+    urls = websites.split("\n")
+
+    # Get all pages
+    results = await up(
+        urls, method="GET", response_fn=lambda x: x.text, raise_for_status=False
+    )
+
+    # Print the content of the first 5 pages
+    for url, content in zip(urls, results[:5]):
+        print(url, repr(content[:100]))
+
+
+results = asyncio.run(main())
diff --git a/docs/examples/wordpress.py b/docs/examples/wordpress.py
index d80281d..fc0a4f4 100644
--- a/docs/examples/wordpress.py
+++ b/docs/examples/wordpress.py
@@ -11,27 +11,44 @@
 import httpx
 
 from unparallel import up
+from unparallel.unparallel import RequestError
 
 
 async def main():
     page_size = 20
-    url = "https://techcrunch.com/wp-json/wp/v2"
-    pagination_url = f"{url}/posts?per_page={page_size}"
+    base_url = "https://techcrunch.com/wp-json/wp/v2"
+    pagination_url = f"/posts?per_page={page_size}"
 
     # Get page count
     page_size = 20
-    response = httpx.head(pagination_url)
+    response = httpx.head(base_url + pagination_url)
     total_pages = int(response.headers["X-WP-TotalPages"])
-    print(f"Website '{url}' has {total_pages} pages (page size = {page_size})")
+    print(f"Website '{base_url}' has {total_pages} pages (page size = {page_size})")
+
+    # Comment the line below to get all pages. Note that you will have to adjust
+    # the settings for this to work without errors. For me, it worked using
+    # max_connections=800 and timeout=180
 
-    # Comment the line below to get all pages
     total_pages = min(total_pages, 1000)
 
-    # Get all pages
+    # Get all pages and flatten the result
     paths = [f"{pagination_url}&page={i}" for i in range(1, total_pages + 1)]
-    return await up(url, paths, method="GET", flatten_result=True)
+    results = await up(paths, method="GET", base_url=base_url, flatten_result=True)
+
+    # Check if some requests failed
+    valid_results = [item for item in results if not isinstance(item, RequestError)]
+    fails = len(results) - len(valid_results)
+    print(f"{fails=} ({fails/len(results):.2%})")
+
+    # Display some properties of the first 5 posts
+    intersting_keys = ["id", "date", "slug", "title", "author"]
+    pp(
+        [
+            {k: v for k, v in item.items() if k in intersting_keys}
+            for item in valid_results[:5]
+        ]
+    )
 
 
-results = asyncio.run(main())
-intersting_keys = ["id", "date", "slug", "title", "author"]
-pp([{k: v for k, v in item.items() if k in intersting_keys} for item in results[:5]])
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/docs/usage.md b/docs/usage.md
index f3dfb5c..b3bf0d9 100644
--- a/docs/usage.md
+++ b/docs/usage.md
@@ -22,7 +22,6 @@ Making async requests: 100%|███████████| 100/100 [00:01<00
 [{'foo': '0'}, {'foo': '1'}, {'foo': '2'}, {'foo': '3'}, {'foo': '4'}]
 ```
 
-
 ## POSTing Data
 ### One body for different paths
 If you want to do POST instead of GET requests, you just have to pass `method='POST'`
@@ -86,16 +85,77 @@ Making async requests: 100%|███████████| 100/100 [00:01<00
  '{"type": "tree", "height": 4}']
 ```
 
-## :construction: Other HTTP methods
+## Custom response functions
+Per default, Unparallel will call the `.json()` function on every `httpx.Response` and 
+return the result. But you might want to get other data from the response like the 
+content as plain-text or the HTTP status. 
+
+For this, define your own response function/callback using a `def` or `lambda` and pass
+it to `up()` via the keyword `response_fn`. The function will receive a `httpx.Response`
+object as the argument and can return anything.
+
+If you want to process the raw `httpx.Response` later yourself, you can simply set 
+`response_fn=None`.
+
+The example below demonstrates how to use a custom response function to get the `.text`
+of a response:
+
+```python hl_lines="11"
+import asyncio
+
+from unparallel import up
 
-!!! warning
+async def main():
+    urls = [
+        "https://www.example.com",
+        "https://duckduckgo.com/",
+        "https://github.com"
+    ]
+    return await up(urls, response_fn=lambda x: x.text)
 
-    Currently (version `0.1.0`) the method `.json()` is called on every response.
-    Hence, HTTP methods that don't return a JSON body will not work as expected.
+results = asyncio.run(main())
+for res in results:
+    print(repr(res[:50]))
+```
+
+This should print something similar to the following:
+```
+Making async requests: 100%|███████████| 3/3 [00:00<00:00,  4.43it/s]
+'<!doctype html>\n<html>\n<head>\n    <title>Example D'
+'<!DOCTYPE html><html lang="en-US"><head><meta char'
+'\n\n\n\n\n\n<!DOCTYPE html>\n<html\n  lang="en"\n  \n  \n  da'
+```
 
+## Other HTTP methods
 Besides the popular GET and POST methods, you can use any other HTTP method supported 
 by HTTPX - which are `GET`, `POST`, `PUT`, `DELETE`, `HEAD`, `PATCH`, and `OPTIONS`.
 
+For example, you can get the status of services/webpages using the method `HEAD` in
+combination with a custom response function:
+
+```python hl_lines="11"
+import asyncio
+
+from unparallel import up
+
+async def main():
+    urls = [
+        "https://www.example.com",
+        "https://duckduckgo.com/",
+        "https://github.com"
+    ]
+    return await up(urls, method="HEAD", response_fn=lambda x: x.status_code)
+
+results = asyncio.run(main())
+print(results)
+```
+
+This prints:
+```
+Making async requests: 100%|███████████| 3/3 [00:00<00:00,  4.43it/s]
+[200, 200, 200]
+```
+
 
 ## Pagination and flattening
 Another use case for Unparallel could be fetching all items from a REST API using 
diff --git a/unparallel/unparallel.py b/unparallel/unparallel.py
index 873e67a..f5972d8 100644
--- a/unparallel/unparallel.py
+++ b/unparallel/unparallel.py
@@ -1,7 +1,7 @@
 import asyncio
 import logging
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 
 import httpx
 from tqdm.asyncio import tqdm as tqdm_async
@@ -9,6 +9,7 @@
 logger = logging.getLogger(__name__)
 VALID_HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "HEAD", "PATCH", "OPTIONS")
 
+DEFAULT_JSON_FN = httpx.Response.json
 DEFAULT_TIMEOUT = httpx.Timeout(timeout=10)
 DEFAULT_LIMITS = httpx.Limits(max_connections=100, max_keepalive_connections=20)
 
@@ -45,7 +46,9 @@ async def single_request(
     url: str,
     method: str,
     json: Optional[Any] = None,
+    response_fn: Optional[Callable[[httpx.Response], Any]] = DEFAULT_JSON_FN,
     max_retries_on_timeout: int = 3,
+    raise_for_status: bool = True,
 ) -> Tuple[int, Any]:
     """Do a single web request for the given URL, HTTP method, and playload.
 
@@ -55,8 +58,12 @@ async def single_request(
         url (str): The URL after the base URI.
         method (str): The HTTP method.
         json (Optional[Any], optional): The JSON payload. Defaults to None.
+        response_fn (Optional[Callable[[httpx.Response], Any]]): The function to apply
+            on every response of the HTTP requests. Defaults to ``httpx.Response.json``.
         max_retries_on_timeout (int): The maximum number retries if the requests fails
             due to a timeout (``httpx.TimeoutException``). Defauls to 3.
+        raise_for_status (bool): If True, ``.raise_for_status()`` is called on the
+            response.
 
     Returns:
         Tuple[int, Any]: A tuple of the index and the JSON response.
@@ -66,9 +73,12 @@ async def single_request(
     for trial in range(1, max_retries_on_timeout + 1):
         try:
             response = await client.request(method, url, json=json)
-            response.raise_for_status()
-            json_data = response.json()
-            return idx, json_data
+            if raise_for_status:
+                response.raise_for_status()
+            if response_fn is None:
+                return idx, response
+            result = response_fn(response)
+            return idx, result
         except httpx.TimeoutException as timeout_ex:
             exception = timeout_ex
             await asyncio.sleep(1)
@@ -98,8 +108,10 @@ async def request_urls(
     base_url: Optional[str] = None,
     headers: Optional[Dict[str, Any]] = None,
     payloads: Optional[Any] = None,
+    response_fn: Optional[Callable[[httpx.Response], Any]] = DEFAULT_JSON_FN,
     flatten_result: bool = False,
     max_retries_on_timeout: int = 3,
+    raise_for_status: bool = True,
     limits: httpx.Limits = DEFAULT_LIMITS,
     timeouts: httpx.Timeout = DEFAULT_TIMEOUT,
     progress: bool = True,
@@ -113,10 +125,14 @@ async def request_urls(
         method (str): HTTP method to use, e.g. get, post, etc.
         base_url (Optional[str]): The base URL of the service, e.g. http://localhost:8000.
         headers (Optional[Dict[str, Any]]): A dictionary of headers to use.
-        payloads (Optional[Any]): A list of JSON payloads (dictionaries) for e.g. HTTP post requests.
-            Used together with ``urls``.
-        flatten_result (bool): If True and the response per request is a list, flatten that
-            list of lists. This is useful when using paging.
+        payloads (Optional[Any]): A list of JSON payloads (dictionaries) for e.g. HTTP
+            post requests. Used together with ``urls``.
+        response_fn (Optional[Callable[[httpx.Response], Any]]): The function to apply
+            on every response of the HTTP requests. Defaults to ``httpx.Response.json``.
+        flatten_result (bool): If True and the response per request is a list, flatten
+            that list of lists. This is useful when using paging.
+        raise_for_status (bool): If True, ``.raise_for_status()`` is called on every
+            response.
         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``.
@@ -145,7 +161,9 @@ async def request_urls(
                     client=client,
                     method=method,
                     json=payloads[i] if payloads else None,
+                    response_fn=response_fn,
                     max_retries_on_timeout=max_retries_on_timeout,
+                    raise_for_status=raise_for_status,
                 )
             )
             tasks.append(task)
@@ -158,7 +176,11 @@ async def request_urls(
 
     results = sort_by_idx(results)
     if flatten_result:
-        return [item for sublist in results for item in sublist]
+        return [
+            item
+            for sublist in results
+            for item in ((sublist,) if isinstance(sublist, RequestError) else sublist)
+        ]
     return results
 
 
@@ -168,10 +190,12 @@ async def up(
     base_url: Optional[str] = None,
     headers: Optional[Dict[str, Any]] = None,
     payloads: Optional[Any] = None,
+    response_fn: Optional[Callable[[httpx.Response], Any]] = DEFAULT_JSON_FN,
     flatten_result: bool = False,
     max_connections: Optional[int] = 100,
     timeout: Optional[int] = 10,
     max_retries_on_timeout: int = 3,
+    raise_for_status: bool = True,
     limits: Optional[httpx.Limits] = None,
     timeouts: Optional[httpx.Timeout] = None,
     progress: bool = True,
@@ -195,6 +219,12 @@ async def up(
             for HTTP post requests. Used together with ``urls``. If one payload but
             multiple URLs are supplied, that payload is used for all requests.
             Defaults to None.
+        response_fn (Optional[Callable[[httpx.Response], Any]]): The function (callback)
+            to apply on every response of the HTTP requests. This can be an existing
+            function of ``httpx.Response`` like ``.json()`` or ``.read()``, or a custom
+            function which takes the ``httpx.Response`` as the argument returns ``Any``.
+            If you set this to ``None``, you will get the raw ``httpx.Response``.
+            Defaults to ``httpx.Response.json``.
         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.
@@ -204,6 +234,8 @@ async def up(
             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.
+        raise_for_status (bool): If True, ``.raise_for_status()`` is called on overy
+            response.
         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``.
@@ -262,8 +294,10 @@ async def up(
         base_url=base_url,
         headers=headers,
         payloads=payloads,
+        response_fn=response_fn,
         flatten_result=flatten_result,
         max_retries_on_timeout=max_retries_on_timeout,
+        raise_for_status=raise_for_status,
         progress=progress,
         limits=limits,
         timeouts=timeouts,


Test patch:
diff --git a/tests/test_unparallel.py b/tests/test_unparallel.py
index c422469..10b1e98 100644
--- a/tests/test_unparallel.py
+++ b/tests/test_unparallel.py
@@ -40,17 +40,55 @@ async def test_single_request(url, method, payload, respx_mock):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("status", [500, 404])
-async def test_single_request_fail(status, respx_mock):
+@pytest.mark.parametrize(
+    "method, return_value, response_fn, expected",
+    [
+        ("HEAD", Response(200), lambda x: x.status_code, 200),
+        ("GET", Response(200, text="foobar"), lambda x: x.text, "foobar"),
+        ("POST", Response(200), None, Response(200)),
+    ],
+    ids=["HEAD_status", "GET_text", "POST_raw"],
+)
+async def test_single_request_custom_response(
+    respx_mock, method, return_value, response_fn, expected
+):
+    url = "http://example.com"
+    respx_mock.request(method, url).mock(return_value=return_value)
+
+    async with AsyncClient() as session:
+        _, result = await single_request(
+            1, session, url, method=method, response_fn=response_fn
+        )
+
+    if isinstance(expected, Response):
+        assert isinstance(result, Response)
+        assert result.status_code == expected.status_code
+    else:
+        assert result == expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "status, do_raise", [(500, True), (500, False), (404, True), (404, False)]
+)
+async def test_single_request_fail(status, do_raise, respx_mock):
     url = "http://test.com/foo"
     respx_mock.get(url).mock(return_value=Response(status))
     session = AsyncClient()
-    idx, result = await single_request(1, session, url=url, method="GET")
+    idx, result = await single_request(
+        1, session, url=url, method="GET", raise_for_status=do_raise, response_fn=None
+    )
+
     assert idx == 1
-    assert isinstance(result, RequestError)
-    assert result.method == "GET"
-    assert result.url == url
-    assert isinstance(result.exception, Exception)
+
+    if do_raise:
+        assert isinstance(result, RequestError)
+        assert result.method == "GET"
+        assert result.url == url
+        assert isinstance(result.exception, Exception)
+    else:
+        assert isinstance(result, Response)
+
     await session.aclose()
 
 
@@ -148,6 +186,17 @@ async def test_request_urls_flat(patched_fetch, flatten, expected):
     assert results == expected
 
 
+@pytest.mark.asyncio
+async def test_up_custom_response_text(respx_mock):
+    urls = ["http://test.com", "http://example.com"]
+    for url in urls:
+        respx_mock.get(url).mock(return_value=Response(200, text="foobar"))
+
+    results = await up(urls, method="GET", response_fn=lambda x: x.text)
+
+    assert results == ["foobar", "foobar"]
+
+
 @pytest.mark.asyncio
 async def test_up_misaligned_paths_and_payloads():
     with pytest.raises(ValueError):