Async http client

.gitignore

@@ -139,4 +139,7 @@ dmypy.json
 .idea/
 
 # MacOS files
-.DS_Store
+.DS_Store
+
+# Local test script
+/docs/examples/test.py

README.md

@@ -203,4 +203,74 @@ class KBCStorageClient(HttpClient):
 cl = KBCStorageClient("my_token")
 
 print(cl.get_files())
-```
+```
+
+## Async Usage
+
+The package also provides an asynchronous version of the HTTP client called AsyncHttpClient. 
+It allows you to make asynchronous requests using async/await syntax. To use the AsyncHttpClient, import it from keboola.http_client_async:
+
+```python
+from keboola.http_client import AsyncHttpClient
+```
+
+The AsyncHttpClient class provides similar functionality as the HttpClient class, but with asynchronous methods such as get, post, put, patch, and delete that return awaitable coroutines. 
+You can use these methods within async functions to perform non-blocking HTTP requests.
+
+```python
+import asyncio
+from keboola.http_client import AsyncHttpClient
+
+async def main():
+    base_url = "https://api.example.com/"
+    async with AsyncHttpClient(base_url) as client:
+        response = await client.get("endpoint")
+
+        if response.status_code == 200:
+            data = response.json()
+            # Process the response data
+        else:
+            # Handle the error
+
+asyncio.run(main())
+```
+
+The AsyncHttpClient provides similar initialization and request methods as the HttpClient. 
+The request methods return awaitable coroutines that can be awaited in an asynchronous context.
+
+#### Building HTTP client based on AsyncHttpClient Example
+This example demonstrates the default use of the HTTPClient as a base for REST API clients.
+
+```python
+import asyncio
+from keboola.http_client import AsyncHttpClient
+
+BASE_URL = 'https://connection.keboola.com/v2/storage'
+MAX_RETRIES = 3
+
+class KBCStorageClient(AsyncHttpClient):
+
+    def __init__(self, storage_token):
+        AsyncHttpClient.__init__(
+            self,
+            base_url=BASE_URL,
+            retries=MAX_RETRIES,
+            backoff_factor=0.3,
+            retry_status_codes=[429, 500, 502, 504],
+            auth_header={"X-StorageApi-Token": storage_token}
+        )
+
+    async def get_files(self, show_expired=False):
+        params = {"showExpired": show_expired}
+        response = await self.get('tables', params=params, timeout=5)
+        return response
+
+async def main():
+    cl = KBCStorageClient("my_token")
+    files = await cl.get_files(show_expired=False)
+    print(files)
+
+asyncio.run(main())
+```
+**Note:** Since there are no parallel requests being made, you won't notice any speedup for this use case.
+For an example where you can see the speedup thanks to async requests, you can view the pokeapi.py in docs/examples.

docs/examples/poekapi_async.py

@@ -0,0 +1,61 @@
+import time
+import asyncio
+from keboola.http_client import AsyncHttpClient
+import csv
+import httpx
+import os
+
+
+async def fetch_pokemon(client, poke_id):
+    try:
+        r = await client.get(str(poke_id))
+        return r
+    except httpx.HTTPStatusError as e:
+        if e.response.status_code == 404:
+            return None
+        else:
+            raise
+
+
+async def save_to_csv(details):
+    filename = "pokemon_details.csv"
+    fieldnames = ["name", "height", "weight"]
+
+    file_exists = os.path.isfile(filename)
+    mode = "a" if file_exists else "w"
+
+    with open(filename, mode, newline="") as csvfile:
+        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
+
+        if not file_exists:
+            writer.writeheader()
+
+        writer.writerow({
+            "name": details["name"],
+            "height": details["height"],
+            "weight": details["weight"]
+        })
+
+
+async def main_async():
+    base_url = "https://pokeapi.co/api/v2/pokemon/"
+    start_time = time.time()
+
+    async with AsyncHttpClient(base_url=base_url, max_requests_per_second=20) as c:
+        poke_id = 1
+
+        while True:
+            details = await fetch_pokemon(c, poke_id)
+            if details is None:
+                break
+
+            await save_to_csv(details)
+
+            poke_id += 1
+
+    end_time = time.time()
+    print(f"Async: Fetched details for {poke_id - 1} Pokémon in {end_time - start_time:.2f} seconds.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main_async())

docs/examples/pokeapi_process_multiple.py

@@ -0,0 +1,44 @@
+import asyncio
+import csv
+import time
+from typing import List
+
+from keboola.http_client import AsyncHttpClient
+
+
+def generate_jobs(nr_of_jobs):
+    return [{'method': 'GET', 'endpoint': str(endpoint)} for endpoint in range(1, nr_of_jobs+1)]
+
+def save_to_csv(results: List[dict]):
+    filename = "pokemon_details.csv"
+    fieldnames = ["name", "height", "weight"]  # Define the fields you want to store
+
+    with open(filename, "w", newline="") as csvfile:
+        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
+        writer.writeheader()
+        for result in results:
+            writer.writerow({
+                "name": result["name"],
+                "height": result["height"],
+                "weight": result["weight"]
+            })
+
+async def main_async():
+    base_url = "https://pokeapi.co/api/v2/pokemon/"
+    start_time = time.time()
+
+    client = AsyncHttpClient(base_url=base_url, max_requests_per_second=20)
+
+    jobs = generate_jobs(1000)
+
+    results = await client.process_multiple(jobs)
+    await client.close()
+
+    end_time = time.time()
+    print(f"Fetched details for {len(results)} Pokémon in {end_time - start_time:.2f} seconds.")
+
+    save_to_csv(results)
+
+
+if __name__ == "__main__":
+    asyncio.run(main_async())

docs/examples/storage_client.py

@@ -0,0 +1,20 @@
+from keboola.http_client import HttpClient
+
+BASE_URL = 'https://connection.keboola.com/v2/storage'
+MAX_RETRIES = 3
+
+
+class KBCStorageClient(HttpClient):
+
+    def __init__(self, storage_token):
+        HttpClient.__init__(self, base_url=BASE_URL, max_retries=MAX_RETRIES, backoff_factor=0.3,
+                            status_forcelist=(429, 500, 502, 504),
+                            default_http_header={"X-StorageApi-Token": storage_token})
+
+    def get_files(self, show_expired=None):
+        params = {"include": show_expired}
+        return self.get('tables', params=params, timeout=5)
+
+cl = KBCStorageClient("my_token")
+
+print(cl.get_files())

docs/examples/storage_client_async.py

@@ -0,0 +1,28 @@
+import asyncio
+from keboola.http_client import AsyncHttpClient
+
+BASE_URL = 'https://connection.keboola.com/v2/storage'
+MAX_RETRIES = 3
+
+class KBCStorageClient(AsyncHttpClient):
+
+    def __init__(self, storage_token):
+        super().__init__(
+            base_url=BASE_URL,
+            retries=MAX_RETRIES,
+            backoff_factor=0.3,
+            retry_status_codes=[429, 500, 502, 504],
+            auth_header={"X-StorageApi-Token": storage_token}
+        )
+
+    async def get_files(self, show_expired=False):
+        params = {"showExpired": show_expired}
+        response = await self.get('tables', params=params, timeout=5)
+        return response
+
+async def main():
+    cl = KBCStorageClient("my_token")
+    files = await cl.get_files(show_expired=False)
+    print(files)
+
+asyncio.run(main())

requirements.txt

@@ -1 +1,3 @@
-requests
+requests
+httpx==0.27.0
+aiolimiter==1.1.0

setup.py

@@ -16,7 +16,8 @@
     setup_requires=['pytest-runner', 'flake8'],
     tests_require=['pytest'],
     install_requires=[
-        'requests'
+        'requests',
+        'httpx'
     ],
     author_email="support@keboola.com",
     description="General HTTP requests library for Python applications running in Keboola Connection environment",

src/keboola/http_client/__init__.py

@@ -1 +1,2 @@
-from .http import HttpClient  # noqa
+from .client import HttpClient  # noqa
+from .async_client import AsyncHttpClient # noqa