from sqlalchemy import create_engine, event, text
from azure.identity import DefaultAzureCredential
import struct
import pyodbc
try:
azure_credentials = DefaultAzureCredential()
TOKEN_URL = "https://database.windows.net/"
SQL_COPT_SS_ACCESS_TOKEN = 1256
server = 'azsql-mi-xyz.database.windows.net'
port = "1234"
database = 'database'
driver = "ODBC Driver 18 for SQL Server"
conn_str = f"mssql+pyodbc://@{server}:{port}/{database}?driver={driver}"
engine = create_engine(conn_str)
@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
# Remove "Trusted_Connection" parameter added by SQLAlchemy
cargs[0] = cargs[0].replace(";Trusted_Connection=Yes", "")
# Generate token credential
raw_token = azure_credentials.get_token(TOKEN_URL).token.encode("utf-16-le")
token_struct = struct.pack(f"<I{len(raw_token)}s", len(raw_token), raw_token)
# Apply it to connection parameters
cparams["attrs_before"] = {SQL_COPT_SS_ACCESS_TOKEN: token_struct}
# Test the connection
with engine.connect() as connection:
query = text("SELECT * FROM tablex")
rows = connection.execute(query).fetchall()
data = [dict(row._mapping) for row in rows]
return {"data": data}
except Exception as e:
raise HTTPException(status_code=500, detail=f"General Error: {str(e)}")
BELOW是来自应用程序的错误消息
thermer错误：DefaultazureCredential未能从随附的凭证中检索令牌。 尝试的凭证：
环境压缩：不可用的环境认证。环境变量尚未完全配置。 请访问
Https：//aka.ms/azsdk/python/istentity/environmentcredential/troubleshoot以解决此问题。 管理IntectidentityCredential：不可用的管理认证验证，IMDS端点没有响应。 共享Tokencachecredential：不可用的sharedTokencachecredential身份验证。在缓存中没有发现帐户。 Azurecrecredential：在路径上找不到Azure CLI Azurepowershellcredential：未安装PowerShellAzuredeplovenerClicential：Azure开发人员CLI找不到。请访问
Https：//aka.ms/azure-dev
以获取安装说明，然后使用“ AZD Auth login”。 要减轻此问题，请参考此处的故障排除指南https：//aka.ms/azsdk/python/indentity/defaultazurecrecredential/troubleshoot.
试图按以下顺序获取凭据，一旦获得了任何凭据，它就会停止。
环环境
WorkloadIdentityCredential
ManagedIdentityCredential
VisualStudiocredential
VisualStudiocodeCredential