Allow to attach sessionmaker after configuring starlette app

[kamilglod]

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 app

Describe 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

[Vasiliy566]

Have you tried to store db instance in fastapi app state? following this way you can easily get wrapper instance in admin

[kamilglod]

@Vasiliy566 I don't follow. Could you share some example code?

[aminalaee]

@kamilglod I think the change should be really simple, we have to refactor this:
https://github.com/aminalaee/sqladmin/blob/d99817aa25edf9039e381f42f5ea2a351457e012/sqladmin/application.py#L84C21-L89C85
I will take a look when I get a chance, or feel free to pick it up.