我想要通过 SQLAlchemy 使用 AD 令牌连接到 Azure SQL DB。
我已经关注了https://github.com/felipefandrade/azuresqlspn并且可以通过此方法成功连接。但是，我想扩展它并使用 SQLAlchemy 来管理连接。
from os import environ
import struct
import adal
from sqlalchemy import create_engine
import pyodbc
clientSecret = environ.get('clientSecret')
clientID = environ.get('clientID')
tenantID = environ.get('tenantID')
authorityHostUrl = "https://login.microsoftonline.com"
authority_url = authorityHostUrl + '/' + tenantID
resource = "https://database.windows.net/"
context = adal.AuthenticationContext(authority_url, api_version=None)
token = context.acquire_token_with_client_credentials(
resource,
clientID,
clientSecret)
tokenb = bytes(token["accessToken"], "UTF-8")
exptoken = b''
for i in tokenb:
exptoken += bytes({i})
exptoken += bytes(1)
tokenstruct = struct.pack("=i", len(exptoken)) + exptoken
driver = "Driver={ODBC Driver 17 for SQL Server}"
server = ";SERVER={0}".format(environ.get('server'))
database = ";DATABASE={0}".format(environ.get('database'))
connString = driver + server + database
SQL_COPT_SS_ACCESS_TOKEN = 1256
conn = pyodbc.connect(connString, attrs_before={SQL_COPT_SS_ACCESS_TOKEN:tokenstruct})
cursor = conn.cursor()
cursor.execute("SELECT TOP 20 ID, Name FROM [Table1]")
row = cursor.fetchone()
print("######")
print("Using pyodbc directly - This works")
while row:
print (str(row[0]) + " " + str(row[1]))
row = cursor.fetchone()
print("#####")
print("Connecting via sqlalchemy - This doesn't work")
SAconnString = "mssql+pyodbc://<server>.database.windows.net/<database>?driver=ODBC+Driver+17+for+SQL+Server"
db = create_engine(SAconnString, connect_args={'attrs_before': {SQL_COPT_SS_ACCESS_TOKEN:tokenstruct}})
SAconn = db.connect()
result = SAconn.execute("SELECT TOP 20 ID, Name FROM [Table1]")
for row in result:
print(row['ID'] + " " + row['Name'])
如代码中所述，使用pyodbc.connect()方法会起作用。但是，使用 SQLAlchemy 我收到错误FA005] [Microsoft][ODBC Driver 17 for SQL Server]Cannot use Access Token with any of the following options: Authentication, Integrated Security, User, Password。在我看来connect_args选项正在传递令牌，但也有其他选项。如何让 SQLAlchemy 将 AD 令牌正确传递给 pyodbc？
需要显式发送 PyODBC 连接字符串urllib.
connString = driver + server + database
params = urllib.parse.quote(connString)
db = create_engine("mssql+pyodbc:///?odbc_connect={0}".format(params), connect_args={'attrs_before': {SQL_COPT_SS_ACCESS_TOKEN:tokenstruct}})
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
