I am using fastapi very recently and as an exercise I want to connect my fastapi API with a validation service on other server... but I do not know how to do this, I have not found something that will help me in the official documentation.. Will I have to do it with python code? Or is there a way?
1
3 Answers
The accepted answer certainly works, but it is not an effective solution. With each request, the ClientSession is closed, so we lose the advantage [0] of ClientSession: connection pooling, keepalives, etc. etc.
We can use the startup and shutdown events [1] in FastAPI, which are triggered when the server starts and shuts down respectively. In these events it is possible to create a ClientSession instance and use it during the runtime of the whole application (and therefore utilize its full potential).
The ClientSession instance is stored in the application state. [2]
Here I answered a very similar question in the context of the aiohttp server: https://stackoverflow.com/a/60850857/752142
from __future__ import annotations
import asyncio
from typing import Final
from aiohttp import ClientSession
from fastapi import Depends, FastAPI
from starlette.requests import Request
app: Final = FastAPI()
@app.on_event("startup")
async def startup_event():
setattr(app.state, "client_session", ClientSession(raise_for_status=True))
@app.on_event("shutdown")
async def shutdown_event():
await asyncio.wait((app.state.client_session.close()), timeout=5.0)
def client_session_dep(request: Request) -> ClientSession:
return request.app.state.client_session
@app.get("/")
async def root(
client_session: ClientSession = Depends(client_session_dep),
) -> str:
async with client_session.get(
"https://example.com/", raise_for_status=True
) as the_response:
return await the_response.text()
Comments
You will need to code it with Python.
If you're using async you should use a HTTP client that is also async, for example aiohttp.
import aiohttp
@app.get("/")
async def slow_route():
async with aiohttp.ClientSession() as session:
async with session.get("http://validation_service.com") as resp:
data = await resp.text()
# do something with data
Comments
The startup and shutdown event handling is deprecated (See here).
But you could use lifespan now:
import asyncio
from contextlib import asynccontextmanager
from fastapi import Request
from httpx import AsyncClient
@asynccontextmanager
async def lifespan(app: FastAPI):
setattr(app.state, "client_session", AsyncClient())
yield
await app.state.client_session.aclose()
def client_session_dep(request: Request) -> AsyncClient:
return request.app.state.client_session
@app.get("/")
async def root(
client_session: AsyncClient = Depends(client_session_dep),
) -> str:
response = await client_session.get('https://www.example.com/')
return response.text()
requestslibrary