Compartir a través de

Compilación de una aplicación de consola de Python con Azure DocumentDB

Esta guía le guía a través de la creación de una aplicación de consola de Python para conectarse a un clúster de Azure DocumentDB. Configure el entorno de desarrollo, autentíquese mediante el azure.identity paquete del SDK de Azure para Python y realice operaciones como crear, consultar y actualizar documentos.

Prerrequisitos

  • Una suscripción de Azure

  • Un clúster de Azure DocumentDB existente

  • La autenticación de Microsoft Entra está configurada para el clúster con el rol root concedido a la identidad.

  • Versión más reciente de Python.

Configuración de la aplicación de consola

A continuación, cree un nuevo proyecto de aplicación de consola e importe las bibliotecas necesarias para autenticarse en el clúster.

  1. Cree un nuevo directorio para el proyecto y configure un entorno virtual.

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

  2. Active el entorno virtual.

    # On Windows
.venv\Scripts\activate

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

  3. Cree un nuevo archivo de Python para la aplicación.

    touch app.py

  4. Instale la biblioteca para la azure.identity autenticación de Azure.

    pip install azure.identity

  5. Instale el pymongo controlador para Python.

    pip install pymongo

Conectarse al clúster

Ahora, utilice la Azure.Identity biblioteca para obtener un TokenCredential que usará para conectarse a su clúster. El controlador oficial de MongoDB tiene una interfaz especial que se debe implementar para obtener tokens de Microsoft Entra para su uso al conectarse al clúster.

  1. Importe los módulos necesarios en la parte superior del archivo de Python.

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

  2. Cree una clase personalizada que implemente la interfaz de devolución de llamada de OpenID Connect (OIDC) de MongoDB.

    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. Establezca la variable de nombre del clúster.

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

  4. Cree una instancia de DefaultAzureCredential y configure las propiedades de autenticación.

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

  5. Cree un cliente de MongoDB configurado con la autenticación de 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")

Realización de operaciones comunes

Por último, use la biblioteca oficial para realizar tareas comunes con bases de datos, colecciones y documentos. Aquí, usará las mismas clases y métodos que usaría para interactuar con MongoDB o DocumentDB para administrar las colecciones y los elementos.

  1. Obtenga una referencia a la base de datos.

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

print("Database pointer created")

  2. Obtenga una referencia a tu colección.

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

print("Collection pointer created")

  3. Cree un documento y actualice/inserte en la colección con collection.update_one.

    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. Use collection.find_one para recuperar un documento específico de la colección.

    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. Consulta varios documentos con collection.find que coinciden con un filtro.

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

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

Contenido relacionado

  • Last updated on