(app: fastapi.FastAPI)
| 33 | |
| 34 | @asynccontextmanager |
| 35 | async def lifespan(app: fastapi.FastAPI) -> AsyncIterator[State]: |
| 36 | context = await common_parameters() |
| 37 | azure_credential = None |
| 38 | if ( |
| 39 | os.getenv("OPENAI_CHAT_HOST") == "azure" |
| 40 | or os.getenv("OPENAI_EMBED_HOST") == "azure" |
| 41 | or os.getenv("POSTGRES_HOST", "").endswith(".database.azure.com") |
| 42 | ): |
| 43 | azure_credential = await get_azure_credential() |
| 44 | engine = await create_postgres_engine_from_env(azure_credential) |
| 45 | sessionmaker = await create_async_sessionmaker(engine) |
| 46 | chat_client = await create_openai_chat_client(azure_credential) |
| 47 | embed_client = await create_openai_embed_client(azure_credential) |
| 48 | if os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING"): |
| 49 | SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine) |
| 50 | yield {"sessionmaker": sessionmaker, "context": context, "chat_client": chat_client, "embed_client": embed_client} |
| 51 | await engine.dispose() |
| 52 | |
| 53 | |
| 54 | def create_app(testing: bool = False): |
nothing calls this directly
no test coverage detected