Skip to content

Instantly share code, notes, and snippets.

@akhundMurad
Last active September 27, 2022 18:27
Show Gist options
  • Select an option

  • Save akhundMurad/6da3d530ed32a674ed500eacc4b6be4d to your computer and use it in GitHub Desktop.

Select an option

Save akhundMurad/6da3d530ed32a674ed500eacc4b6be4d to your computer and use it in GitHub Desktop.
Aiohttp client session and FastAPI DI
from di import ClientSessionManager, provide_client_session_stub, provide_client_session
async def on_shutdown() -> None:
await ClientSessionManager.close()
def build_asgi_application() -> FastAPI:
application = FastAPI(on_shutdown=[on_shutdown])
app.dependency_overrides[provide_client_session_stub] = provide_client_session
return app
import aiohttp
class ClientSessionManager:
_client_session: aiohttp.ClientSession | None = None
@property
def client_session(self) -> aiohttp.ClientSession:
return self.build()
@classmethod
def build(cls) -> aiohttp.ClientSession:
if not cls._client_session:
cls._client_session = aiohttp.ClientSession()
return cls._client_session
@classmethod
async def close(cls) -> None:
if cls._client_session:
await cls._client_session.close()
cls._client_session = None
def provide_client_session_stub():
...
def provide_client_session() -> aiohttp.ClientSession:
return ClientSessionManager.client_session
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment