Last active
September 27, 2022 18:27
-
-
Save akhundMurad/6da3d530ed32a674ed500eacc4b6be4d to your computer and use it in GitHub Desktop.
Aiohttp client session and FastAPI DI
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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