با استفاده از توکن های Entra ID به پایگاه داده Azure SQL در SQL Alchemy متصل شوید

Summarize this content to 400 words in Persian Lang
ما یک برنامه وب در حال کار داریم که باید برای مدت زمان درخواست به پایگاه داده Azure SQL ما متصل شود. برای سهولت پرس و جو در پایگاه داده، از SQL Alchemy و pyodbc.

ما چند هدف داریم که می خواهیم به آنها برسیم:

هر درخواست جلسه مخصوص به خود را دارد. ما می خواهیم با شروع درخواست، یک جلسه جدید باز کنیم و پس از انجام آن، آن را ببندیم.
ما می خواهیم با استفاده از هویت مدیریت شده وب سرویس (Azure Function App یا Azure Web App) به پایگاه داده متصل شویم.
ما می خواهیم تا جایی که می توانیم رسیدگی به جلسه و اتصال را محول کنیم.

مدیریت طول عمر جلسه

ابتدا، برای اطمینان از باز کردن یک جلسه جدید برای هر درخواست، می‌توانیم توابع درخواست را در دکوراتورها قرار دهیم تا مطمئن شوند که یک جلسه جدید ایجاد شده و متعاقباً از بین می‌رود.

برای اطمینان از اینکه جلسه ما به راحتی از سراسر برنامه بدون نیاز به تحویل دادن آن در هر فراخوانی قابل دسترسی است، از یک الگوی singleton استفاده می کنیم. با این حال، از آنجایی که ما چندین رشته را برای رسیدگی به چندین درخواست به طور همزمان اجرا می کنیم، باید مطمئن شویم که هیچ شرایط مسابقه ای در مورد اشیاء جلسه وجود ندارد.

SQL Alchemy یک ابزار عالی برای آسان کردن این کار دارد: جلسه scoped.

برای استفاده از این، ما کارخانه اتصال خود را در یک بسته بندی می کنیم scoped_session() تماس بگیرید:

from urllib.parse import quote_plus
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

connection_string = “…”
engine = create_engine(“mssql+pyodbc:///?odbc_connect={}”.format(quote_plus(connection_string))
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

حالا هر وقت بخواهیم از جلسه استفاده کنیم فقط تماس می گیریم Session() و SQL Alchemy اطمینان حاصل می کند که ما از شی جلسه موجود دوباره استفاده می کنیم. اگر کارمان تمام شد، می توانیم تماس بگیریم Session.remove() برای بستن جلسه هیچ راهی برای SQL Alchemy وجود ندارد که بگوید آیا موضوع انجام شده است، بنابراین ما باید خودمان این کار را انجام دهیم.

از طریق نشانه های Entra ID با استفاده از هویت مدیریت شده متصل شوید

چند پست وجود دارد که نحوه راه اندازی اتصال به پایگاه های داده Azure SQL با استفاده از نشانه های دسترسی را توضیح می دهد، اما بهترین منبع برای این کار خود اسناد SQL Alchemy است.

بیایید با هم به جزئیات بپردازیم. ابتدا به یک رشته اتصال نیاز داریم. از آنجایی که می خواهیم به هویت مدیریت شده (یا Azure CLI برای توسعه محلی) تکیه کنیم، هیچ اعتباری را در رشته اتصال قرار نمی دهیم:

Driver={ODBC Driver 18 for SQL Server};Database=YOUR_DB;Server=tcp:you.database.windows.net,1433;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30

طرح کلی ما این است:

یک کنترل کننده رویداد اضافه کنید که هر زمان که به پایگاه داده متصل می شویم فعال شود.
هر زمان که به پایگاه داده متصل می شویم، یک نشانه دسترسی را بازیابی کنید.
آرگومان های اتصال را طوری تنظیم کنید که نشانه دسترسی (تازه!) را در رشته اتصال قرار دهیم.

حالا بیایید نگاهی به کد بیندازیم:

import struct
from urllib.parse import quote_plus

from azure.core.exceptions import ClientAuthenticationError
from azure.identity import DefaultAzureCredential
from sqlalchemy import event

@event.listens_for(engine, “do_connect”)
def provide_token(dialect, conn_rec, cargs, cparams):
# Uid may not be provided for the token authentication to work
if “Uid=” in cargs[0]:
return

# Connection option for access tokens as defined in msodbcsql.h
SQL_COPT_SS_ACCESS_TOKEN = 1256

# remove the “Trusted_Connection” parameter that SQLAlchemy adds
cargs[0] = cargs[0].replace(“;Trusted_Connection=Yes”, “”)
cargs[0] = cargs[0].replace(“;Authentication=TokenIdentifiedPrincipal”, “”)

# create token credential => try two ways to make local dev work
scope = “https://database.windows.net/.default”
try:
cred = DefaultAzureCredential(exclude_managed_identity_credential=True).get_token(scope)
except ClientAuthenticationError:
cred = DefaultAzureCredential(
exclude_managed_identity_credential=True,
exclude_shared_token_cache_credential=True,
).get_token(scope)

raw_token = cred.token.encode(“utf-16-le”)
token_struct = struct.pack(f”{len(raw_token)}s”, len(raw_token), raw_token)
cparams[“attrs_before”] = {SQL_COPT_SS_ACCESS_TOKEN: token_struct}

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

خلاصه

با این کار به اهدافمان رسیدیم. با scoped_session() لازم نیست هر بار که درخواستی می آید یک جلسه جدید باز کنیم (این برای ما رسیدگی می شود) اما باید جلسه را در پایان ببندیم تا جلسات بیش از حد آویزان نداشته باشیم.

ما همچنین با استفاده از هویت خود (برای توسعه دهنده محلی) یا هویت مدیریت شده سرویس وب به پایگاه داده SQL متصل می شویم. هر بار که یک اتصال جدید ایجاد می شود، رشته اتصال را تغییر می دهیم.

قدردانی ها

یک تشکر بزرگ از دیوید که به من کمک کرد تا مفهوم جلسات scoped را بفهمم.

from urllib.parse import quote_plus
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

connection_string = "..."
engine = create_engine("mssql+pyodbc:///?odbc_connect={}".format(quote_plus(connection_string))
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

import struct
from urllib.parse import quote_plus

from azure.core.exceptions import ClientAuthenticationError
from azure.identity import DefaultAzureCredential
from sqlalchemy import event


@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
    # Uid may not be provided for the token authentication to work
    if "Uid=" in cargs[0]:
        return

    # Connection option for access tokens as defined in msodbcsql.h
    SQL_COPT_SS_ACCESS_TOKEN = 1256

    # remove the "Trusted_Connection" parameter that SQLAlchemy adds
    cargs[0] = cargs[0].replace(";Trusted_Connection=Yes", "")
    cargs[0] = cargs[0].replace(";Authentication=TokenIdentifiedPrincipal", "")

    # create token credential => try two ways to make local dev work
    scope = "https://database.windows.net/.default"
    try:
        cred = DefaultAzureCredential(exclude_managed_identity_credential=True).get_token(scope)
    except ClientAuthenticationError:
        cred = DefaultAzureCredential(
            exclude_managed_identity_credential=True,
            exclude_shared_token_cache_credential=True,
        ).get_token(scope)

    raw_token = cred.token.encode("utf-16-le")
    token_struct = struct.pack(f"{len(raw_token)}s", len(raw_token), raw_token)
    cparams["attrs_before"] = {SQL_COPT_SS_ACCESS_TOKEN: token_struct}