SqlAlchemy connect to Azure Postgresql Flexible Database with token authentication

Question

I would want to connect SqlAlchemy to Azure Database for Postgres with token based authentication to achieve Single Sign On (SSO) by using Azure Active Directory (AAD), but seems like SqlAlchemy poorly support token-based connection string, is there any way to achieve it? In my solution, SqlAlchemy is required, and Single-Sign-On with Azure AD is required, and no password provided.

I browsed and it seems like pyodbc library work with token auth, pyodbc lib token auth, but I need SqlAlchemy, it's very cutting-edge case, I believe.

Answer 1

The best approach would be to make use of do_connect event hook: https://docs.sqlalchemy.org/en/20/core/engines.html#generating-dynamic-authentication-tokens

Assuming you have defined a function to retrieve AD token: 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()

Answer 2

You can follow the procedure below to connect to an Azure Database for PostgreSQL server using access token authentication:

Use the code below to get the access token for authentication:

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')}")

Use the code below to connect to the Azure Database for PostgreSQL server using SQLAlchemy:

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)

Below is the complete code:

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)

You will get the output as mentioned below: