Dynamically update password

[MarkusDressel]

I'am using a Azure PostgreSql with a managed identity. This means, the password is a bearer token which expires after 4 hours. I need to fetch a new token once expires and pass this to your middleware. Is there a (hacky) way for it ?

Any help, highly appreciated

[h0rn3t]

you can try use custom_engine,

ps not tested just idea

from sqlalchemy.engine.create import create_engine
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy.ext.asyncio.engine import create_async_engine

def create_custom_engine(url, **kwargs):
    class CustomAsyncEngine(AsyncEngine):
        async def _update_token(self):
            # your recv token logic
            new_token = fetch_bearer_token()
            self.url = self.url.set(password=new_token)

        async def connect(self, *args, **kwargs):
            await self._update_token()
            return await super().connect(*args, **kwargs)

    return create_async_engine(url, class_=CustomAsyncEngine, **kwargs)


custom_engine = create_custom_engine(db_url, **engine_args)
SQLAlchemyMiddleware(app, custom_engine=custom_engine)