estoy usando fastapi hace muy poco y como ejercicio quiero conectar mi api de fastapi con un servicio de validacion en otro servidor... pero no se como hacer esto, no he encontrado algo que me ayude en la documentacion oficial .. ¿Tendré que hacerlo con código python? ¿O hay una manera?
gracias por su ayuda y disculpe mi inglés.
Deberá codificarlo con Python.
Si está utilizando async, debe usar un cliente HTTP que también sea asíncrono, por ejemplo, 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 dataLa respuesta aceptada ciertamente funciona, pero no es una solución efectiva. Con cada solicitud, ClientSession se cierra, por lo que perdemos la ventaja [0] de ClientSession : agrupación de conexiones, keepalives, etc., etc.
Podemos usar los eventos de startup y shutdown [1] en FastAPI, que se activan cuando el servidor se inicia y se apaga respectivamente. En estos eventos, es posible crear una instancia de ClientSession y usarla durante el tiempo de ejecución de toda la aplicación (y, por lo tanto, utilizar todo su potencial).
La instancia de ClientSession se almacena en el estado de la aplicación. [2]
Aquí respondí una pregunta muy similar en el contexto del servidor aiohttp: 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()