Лучший способ создания движка с использованием sqlalchemy и удостоверений Azure? ⇐ Python

[Anonymous]

Я пытаюсь создать резервный механизм для аутентификации SQL. В настоящее время мы используем sqlalchemy для create_engine и сеанса, используя имя пользователя и пароль SQL. В документации sqlalchemy я обнаружил, что мы можем создать прослушиватель событий.
После многих игр эта реализация работает. Я могу создать механизм и взаимодействовать с базой данных, тогда как аутентификация выполняется с использованием аутентификации на основе браузера с помощью Azure Identity.
Хотя это работает, у меня такое ощущение, что должно быть лучшее решение для этой проблемы. использование аутентифицированного токена для пользователя и выполнение операций, ограниченных этим пользователем.
Кроме того, мне удалось найти документацию Azure, в которой показан более чистый код, но не используется sqlalchemy: https://learn.microsoft.com/en-us/azure ... Csql-inter
Я пытаюсь сделать это с помощью sqlalchemy как надежной функции, чтобы избежать обмена учетными данными.
import os
import struct
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from azure.identity import InteractiveBrowserCredential

# Constants for Azure SQL authentication
SQL_COPT_SS_ACCESS_TOKEN = 1256 # Connection option for access tokens, as defined in msodbcsql.h
TOKEN_URL = "https://database.windows.net/.default" # The token URL for any Azure SQL database

def create_engine_using_interactive_browser_credential():
"""
Create SQLAlchemy engine using InteractiveBrowserCredential for authentication.
"""
credential = InteractiveBrowserCredential(tenant_id=os.getenv("TENANT_ID")) # Opens a browser for user login
database_name = os.getenv("DB_DATABASE", "default_db")
server_name = os.getenv("DB_SERVER", "default_server")
driver = "+".join(os.getenv("DB_DRIVER", "ODBC Driver 17 for SQL Server").split())

connection_string = (
f"mssql+pyodbc://@{server_name}/{database_name}?driver={driver}"
)

engine = create_engine(connection_string, fast_executemany=True)

# Add event listener to provide access token to the connection following the official sqlalchemy documentation: https://docs.sqlalchemy.org/en/14/diale ... ess-tokens
@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
# Remove the "Trusted_Connection" parameter that SQLAlchemy adds
if "Trusted_Connection=Yes" in cargs[0]:
cargs[0] = cargs[0].replace(";Trusted_Connection=Yes", "")

# Generate access token
token = credential.get_token(TOKEN_URL).token
token_bytes = token.encode("utf-16-le")
token_struct = struct.pack(f"

Подробнее здесь: https://stackoverflow.com/questions/793 ... e-identity