Поделиться через

Создание консольного приложения Python с помощью Azure DocumentDB

В этом руководстве описывается создание консольного приложения Python для подключения к кластеру Azure DocumentDB. Вы настраиваете среду разработки, проходите проверку подлинности с помощью azure.identity пакета из пакета Azure SDK для Python и выполняете такие операции, как создание, запросы и обновление документов.

Предпосылки

  • Существующий кластер Azure DocumentDB

Настройка консольного приложения

Затем создайте проект консольного приложения и импортируйте необходимые библиотеки для проверки подлинности в кластере.

  1. Создайте новый каталог для проекта и настройте виртуальную среду.

    mkdir mongodb-app
cd mongodb-app
python -m venv .venv

  2. Включите виртуальную среду.

    # On Windows
.venv\Scripts\activate

# On macOS/Linux
source .venv/bin/activate

  3. Создайте новый файл Python для приложения.

    touch app.py

  4. Установите библиотеку azure.identity для проверки подлинности Azure.

    pip install azure.identity

  5. pymongo Установите драйвер для Python.

    pip install pymongo

Подключение к кластеру

Теперь используйте библиотеку Azure.Identity, чтобы получить TokenCredential для подключения к вашему кластеру. Официальный драйвер MongoDB имеет специальный интерфейс, который должен быть реализован для получения маркеров из Microsoft Entra для использования при подключении к кластеру.

  1. Импортируйте необходимые модули в верхней части файла Python.

    from azure.identity import DefaultAzureCredential
from pymongo import MongoClient
from pymongo.auth_oidc import OIDCCallback, OIDCCallbackContext, OIDCCallbackResult

  2. Создайте пользовательский класс, реализующий интерфейс обратного вызова MongoDB OpenID Connect (OIDC).

    class AzureIdentityTokenCallback(OIDCCallback):
    def __init__(self, credential):
        self.credential = credential

    def fetch(self, context: OIDCCallbackContext) -> OIDCCallbackResult:
        token = self.credential.get_token(
            "https://ossrdbms-aad.database.windows.net/.default").token
        return OIDCCallbackResult(access_token=token)

  3. Задайте переменную имени кластера.

    clusterName = "<azure-documentdb-cluster-name>"

  4. Создайте экземпляр DefaultAzureCredential и настройте свойства проверки подлинности.

    credential = DefaultAzureCredential()
authProperties = {"OIDC_CALLBACK": AzureIdentityTokenCallback(credential)}

  5. Создайте клиент MongoDB, настроенный с проверкой подлинности Microsoft Entra.

    client = MongoClient(
    f"mongodb+srv://{clusterName}.global.mongocluster.cosmos.azure.com/",
    connectTimeoutMS=120000,
    tls=True,
    retryWrites=True,
    authMechanism="MONGODB-OIDC",
    authMechanismProperties=authProperties
)

print("Client created")

Выполнение распространенных операций

Наконец, используйте официальную библиотеку для выполнения общих задач с базами данных, коллекциями и документами. Здесь вы используете те же классы и методы, которые будут использоваться для взаимодействия с MongoDB или DocumentDB для управления коллекциями и элементами.

  1. Получите ссылку на вашу базу данных.

    database = client.get_database("<database-name>")

print("Database pointer created")

  2. Получите ссылку на коллекцию.

    collection = database.get_collection("<container-name>")

print("Collection pointer created")

  3. Создайте документ и обновите или вставьте его в коллекцию с помощью .

    new_document = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
    "category": "gear-surf-surfboards",
    "name": "Yamba Surfboard",
    "quantity": 12,
    "price": 850.00,
    "clearance": False,
}

filter = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
}
payload = {
    "$set": new_document
}
result = collection.update_one(filter, payload, upsert=True)

  4. Используется collection.find_one для извлечения определенного документа из коллекции.

    filter = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
    "category": "gear-surf-surfboards"
}
existing_document = collection.find_one(filter)
print(f"Read document _id:\t{existing_document['_id']}")

  5. Запрос нескольких документов с collection.find, которые соответствуют фильтру.

    filter = {
    "category": "gear-surf-surfboards"
}
matched_documents = collection.find(filter)

for document in matched_documents:
    print(f"Found document:\t{document}")

Связанный контент

  • Last updated on