SqlAlchemy 通过令牌身份验证连接到 Azure Postgresql 灵活数据库

Question

我想使用

SqlAlchemy

将

Azure Database for Postgres

连接到

token based authentication

以通过使用 Azure Active Directory (AAD) 实现单点登录 (SSO)，但似乎 SqlAlchemy 不太支持基于令牌的连接字符串，有什么办法实现它吗？在我的解决方案中，需要 SqlAlchemy，并且需要使用 Azure AD 单点登录，并且不提供密码。

我浏览了一下，似乎 pyodbc 库可以使用令牌身份验证，pyodbc lib token auth，但我需要 SqlAlchemy，我相信这是非常前沿的案例。

Answer 1

您可以按照以下过程使用访问令牌身份验证连接到 Azure Database for PostgreSQL 服务器：

使用以下代码获取用于身份验证的访问令牌：

import msal

# Define Azure AD settings
client_id = "<clientId>"
client_secret = "<clientSecret>"
tenant_id = "<tenantId>"
scope = ["https://graph.microsoft.com/.default"]
authority = f"https://login.microsoftonline.com/{tenant_id}"

app = msal.ConfidentialClientApplication(
    client_id=client_id,
    client_credential=client_secret,
    authority=authority
)

result = None
try:
    result = app.acquire_token_silent(scope, account=None)
    if not result:
        result = app.acquire_token_for_client(scopes=scope)
except Exception as ex:
    print(f"An error occurred: {str(ex)}")

if "access_token" in result:
    access_token = result["access_token"]
    print(access_token)
else:
    print(f"Token acquisition failed: {result.get('error')}")

使用下面的代码通过 SQLAlchemy 连接到 Azure Database for PostgreSQL 服务器：

from sqlalchemy import create_engine, text
import urllib.parse

# Replace these placeholders with your PostgreSQL server details
username = "<activeDirectoryName>"
hostname = "<serverName>.postgres.database.azure.com"
port = "<port>"
database_name = "<databaseName>"

# Properly percent-encode the password
password_encoded = urllib.parse.quote_plus(access_token)

# Create the database URL
db_url = f"postgresql://{username}:{password_encoded}@{hostname}:{port}/{database_name}"

engine = create_engine(db_url)

with engine.connect() as connection:
    query = text("SELECT usename AS role_name FROM pg_catalog.pg_user ORDER BY role_name desc;")
    result = connection.execute(query)

    for row in result:
        print(row)

以下是完整代码：

import msal
from sqlalchemy import create_engine, text
import urllib.parse

# Define Azure AD settings
client_id = "<clientId>"
client_secret = "<clientSecret>"
tenant_id = "<tenantId>"
scope = ["https://graph.microsoft.com/.default"]
authority = f"https://login.microsoftonline.com/{tenant_id}"

app = msal.ConfidentialClientApplication(
    client_id=client_id,
    client_credential=client_secret,
    authority=authority
)

result = None
try:
    result = app.acquire_token_silent(scope, account=None)
    if not result:
        result = app.acquire_token_for_client(scopes=scope)
except Exception as ex:
    print(f"An error occurred: {str(ex)}")

if "access_token" in result:
    access_token = result["access_token"]
else:
    print(f"Token acquisition failed: {result.get('error')}")

# Replace these placeholders with your PostgreSQL server details
username = "<activeDirectoryAdmin>"
hostname = "<serverName>.postgres.database.azure.com"
port = "<port>"
database_name = "<database>"

# Properly percent-encode the password
password_encoded = urllib.parse.quote_plus(access_token)

# Create the database URL
db_url = f"postgresql://{username}:{password_encoded}@{hostname}:{port}/{database_name}"

engine = create_engine(db_url)

with engine.connect() as connection:
    query = text("SELECT usename AS role_name FROM pg_catalog.pg_user ORDER BY role_name desc;")
    result = connection.execute(query)

    for row in result:
        print(row)

您将得到如下输出：

Answer 2

最好的方法是使用

do_connect

事件挂钩：https://docs.sqlalchemy.org/en/20/core/engines.html#generate-dynamic-authentication-tokens

假设您已经定义了一个函数来检索 AD 令牌：

get_authentication_token

:

from sqlalchemy import event

user=<MS Entra principal>
host=<hostname>.postgres.database.azure.com
port=<port>
dbname=<dbname>

engine = create_engine(f"postgresql+psycopg2://{user}@{host}:{port}/{dbname}")

@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
  cparams["password"] = get_authentication_token()