[Python] Refresh a connection token in psycopg3 ConnectionPool

I am using psycopg3 to initialize a AsyncConnectionPool. My Postgres database is hosted in Azure. I am connecting via a managed identity. Therefore I retrieve a token from Microsoft which is valid for one hour. The initializing of the Pool and the first connections (within the hour where the token is valid) works fine. When the token is expired and new connections should be established, I get the following error: "error connecting in 'pool-1': connection failed: connection to server at "127.0.0.1", port 5433 failed: FATAL: The access token has expired. Please acquire a new token and retry."

So I think that the conninfo-string is stored. But somehow I have to call get_conninfo() every time when a new connection should established. Any ideas how to solve this?

Here is some part of my implementation.

@classmethod
async def connect(cls):
"""Initializes the connection pool."""
async with cls._lock:
if cls._instance is None:
cls._instance = cls()
if not hasattr(cls._instance, 'pool'):
# Retrieve connection parameters
cls._instance.user = settings.user
cls._instance.host = settings.host
cls._instance.port = settings.port
cls._instance.dbname = settings.dbname
cls._instance.connection_type = settings.connection_type

if cls._instance.connection_type == "managed_identity":
cls._instance.credential = ManagedIdentityCredential()
else:
cls._instance.password = settings.pgvector.pgvector_password.get_secret_value()

cls._instance.pool = AsyncConnectionPool(
conninfo=await cls._instance.get_conninfo(),
open=True
)
await cls._instance.pool.open()
logger.info("Async database connection pool initialized in PostgresClient.")

@classmethod
async def get_conninfo(cls) -> str:
"""Returns connection info string based on the connection type."""
if cls._instance.connection_type == "managed_identity"
# retrieve token from Microsoft
access_token = cls._instance.credential.get_token(
"https://ossrdbms-aad.database.windows.net/.default"
)
return (
f"dbname={cls._instance.dbname} host={cls._instance.host} "
f"port={cls._instance.port} sslmode=require user={cls._instance.user} "
f"password={access_token.token}"
else:
...

Continue reading...