Aktualisieren Sie ein Verbindungstoken im psycopg3 ConnectionPool ⇐ Python

[Guest]

Ich verwende psycopg3, um einen AsyncConnectionPool zu initialisieren. Meine Postgres-Datenbank wird in Azure gehostet. Ich verbinde mich über eine verwaltete Identität. Daher erhalte ich von Microsoft einen Token, der eine Stunde lang gültig ist. Die Initialisierung des Pools und der ersten Verbindungen (innerhalb der Stunde, in der das Token gültig ist) funktioniert einwandfrei. Wenn das Token abgelaufen ist und neue Verbindungen hergestellt werden sollen, erhalte ich die folgende Fehlermeldung: „Fehler beim Herstellen der Verbindung in ‚Pool-1‘: Verbindung fehlgeschlagen: Verbindung zum Server unter „127.0.0.1“, Port 5433 fehlgeschlagen: FATAL: Das Zugriffstoken.“ ist abgelaufen. Bitte besorgen Sie sich ein neues Token und versuchen Sie es erneut.“
Ich denke also, dass der Conninfo-String gespeichert ist. Aber irgendwie muss ich get_conninfo() jedes Mal aufrufen, wenn eine neue Verbindung hergestellt werden soll. Irgendwelche Ideen, wie man das lösen kann?
Hier ist ein Teil meiner Implementierung.

Code: Select all

    @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:
...