I have this (example) REST API client:
import requests
class FakeHttpClient:
base_url = "https://api.example.com"
def __init__(self) -> None:
self.session = requests.Session()
def get_user_info(self, user_id: str) -> dict:
return self.session.get(f"{self.base_url}/users/{user_id}").json()
def get_users_info(self, user_ids: list[str]) -> list[dict]:
return [self.get_user_info(user_id) for user_id in user_ids]
if __name__ == '__main__':
users = FakeHttpClient().get_users_info(["1", "2", "3"])
I am trying to rewrite this using aiohttp so that the user interface is the same (i.e. creating the API client instance synchronously and calling its methods synchronously)? I have managed to create a fully async client, but this is overkill for users since it is only the get_users_info() method which benefits from asynchronous execution – I don’t want people to have to invoke async and await keywords and use context managers and all that stuff for a single HTTP request which it would be fine to be synchronous.
aiohttp may not be the best fit for your use case of mixing async with sync methods, as it’s specifically designed for asyncio use. Instead, I recommend looking into HTTPX. It has support for both sync and async Clients (similar to Sessions).
Their documentation is easy to get started with. All the best.
An alternative would be to have your aiohttp with the asynchronous operations and to build a wrap on top of it. The final interface is what your users can call synchronously, and this will handle using the client inside the calls asynchronous.
In other words this way you can benefit from asynchronous execution and still avoid users to have to invoke async and await keywords and use context managers and all that stuff.
If we have a simple aynchronous client as:
import aiohttp
import asyncio
class AsyncHttpClient:
base_url = "https://api.example.com"
# base_url = "http://127.0.0.1:5000" # for testing
def __init__(self) -> None:
self.session = None
async def _init_session(self):
if self.session is None:
self.session = aiohttp.ClientSession()
async def get_user_info(self, user_id: str) -> dict:
await self._init_session()
async with self.session.get(f"{self.base_url}/users/{user_id}") as response:
return await response.json()
async def get_users_info(self, user_ids: list[str]) -> list[dict]:
await self._init_session()
tasks = [self.get_user_info(user_id) for user_id in user_ids]
return await asyncio.gather(*tasks)
async def close(self):
if self.session is not None:
await self.session.close()
This client could be wrap into another interface that would be what your users use.
class FakeHttpClient:
def __init__(self) -> None:
self.async_client = AsyncHttpClient()
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
def get_user_info(self, user_id: str) -> dict:
return self.loop.run_until_complete(self.async_client.get_user_info(user_id))
def get_users_info(self, user_ids: list[str]) -> list[dict]:
return self.loop.run_until_complete(self.async_client.get_users_info(user_ids))
def close(self):
self.loop.run_until_complete(self.async_client.close())
self.loop.close()
def __del__(self):
self.close()
if __name__ == '__main__':
users = FakeHttpClient() .get_users_info(["1", "2", "3"])
print(users)
Note the __del__ method into the fake client makes sure that the aiohttp.client.ClientSession is closed
For testing the code
from flask import Flask, jsonify
app = Flask(__name__)
# Sample data
users_data = {
"1": {"id": "1", "name": "Alice", "email": "[email protected]"},
"2": {"id": "2", "name": "Bob", "email": "[email protected]"},
"3": {"id": "3", "name": "Charlie", "email": "[email protected]"}
}
@app.route('/users/<user_id>', methods=['GET'])
def get_user(user_id):
print(f"User Id: {user_id}")
user = users_data.get(user_id)
if user:
return jsonify(user)
return jsonify({"error": f"User {user_id} not found"}), 404
if __name__ == '__main__':
app.run(port=5000)