Allow to attach sessionmaker after configuring starlette app
Checklist
- There are no similar issues or pull requests for this yet.
Is your feature related to a problem? Please describe.
I want to setup sqladmin for project in which we're using google IAM auth for PostgreSQL and our function that creates SqlAlchemy engine is async:
from typing import TYPE_CHECKING
from google.cloud.sql.connector import Connector, create_async_connector
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
if TYPE_CHECKING:
from asyncpg import Connection
class AsyncEngineWrapper:
"""
Reflects the interface of the AsyncEngine but have reference to the Cloud SQL
Connector to close it properly when disposing the engine.
"""
def __init__(self, engine: AsyncEngine, connector: Connector):
self.engine = engine
self.connector = connector
def __getattr__(self, attr):
return getattr(self.engine, attr)
async def dispose(self, close: bool = False) -> None:
await self.connector.close_async()
await self.engine.dispose(close)
async def create_cloud_sql_async_engine(
cloud_sql_instance: str,
*,
cloud_sql_user: str,
cloud_sql_database: str,
cloud_sql_password: str | None = None,
enable_iam_auth: bool = True,
**kwargs,
) -> AsyncEngine:
"""
Use Cloud SQL IAM role authentication mechanism
https://cloud.google.com/sql/docs/postgres/iam-authentication
to create new SqlAlchemy async engine.
"""
connector = await create_async_connector()
async def get_conn() -> "Connection":
return await connector.connect_async(
cloud_sql_instance,
"asyncpg",
user=cloud_sql_user,
password=cloud_sql_password,
db=cloud_sql_database,
enable_iam_auth=enable_iam_auth,
)
engine = create_async_engine(
"postgresql+asyncpg://", async_creator=get_conn, **kwargs
)
return AsyncEngineWrapper(engine, connector) # type: ignore[return-value]and now it's quite tricky to get an instance of the engine when creating FastAPI app (and Admin instance).
Similar problem might have someone that share connection pool between app and admin and creates engine inside of the lifespan https://fastapi.tiangolo.com/advanced/events/ - which happens after FastAPI app is created.
Describe the solution you would like.
I would like to have an option to create app with an admin instance without passing engine or sessionmaker yet. and then have an option (like dedicated Admin method) to set proper sessionmaker or engine inside of the lifespan. Something like:
async def lifespan(app: FastAPI):
engine = await create_async_engine(...)
app.state.admin.attach_engine(engine)
yield
await engine.dispose()
def create_app():
app = FastAPI(lifespan=lifespan)
admin = Admin(app)
app.state.admin = admin
@admin.add_view
class UserAdmin(ModelView, model=User):
column_list = [User.id, User.username]
return appDescribe alternatives you considered
I tried to create whole Admin inside of the lifespan but it didn't work. It looks like it's already too late and we need to set it up when creating FastAPI app.
Additional context
No response
How does funding with Polar work?
Pay now to fund the work behind this issue.
Get updates on progress being made.
Maintainer is rewarded once the issue is completed.
Backer
You're funding impactful open source efforts
Contributor
You want to contribute to this effort
Maintainer
You want to get funding like this too