We are deciding whether to integrate aiohttp to our app (FastAPI) but I am having trouble understanding the logic needed to send authenticated requests using headers that have expiration time.
The provider we want to connect to requires us to first authenticate by sending a POST request to an endpoint that returns a key and a signature to use as a header for subsequent requests. The problem is that these values have a very short expiration time and I need to check all requests to see if they are still valid.
We define a class that we use to raise/stop the aiohttp client session and reuse it throughout the application:
import aiohttp
class HttpClient:
session: aiohttp.ClientSession = None
def start(self):
self.session = aiohttp.ClientSession()
async def stop(self):
await self.session.close()
self.session = None
def __call__(self) -> aiohttp.ClientSession:
assert self.session is not None
return self.session
http_client = HttpClient()
Then we make the requests by importing this session:
import aiohttp
from app.core.http_client import http_client
from fastapi import APIRouter, Depends
router = APIRouter()
@router.get("/some-endpoint")
async def some_endpoint(http_client: aiohttp.ClientSession = Depends(http_client)):
url = "http://example.com/"
headers = { ... }
async with http_client.get(url, headers=headers) as response:
return await response.json()
How can I check if the headers are correct each time and, in case they have expired, get them again and resend the requests?
Is there a way to create a middleware for the client session that runs on every requests? Thanks in advance!