Транзакционные пакетные операции в Azure Cosmos DB

Для создания транзакционной пакетной операции начните с экземпляра контейнера и вызовите CreateTransactionalBatch:

PartitionKey partitionKey = new PartitionKey("road-bikes");

TransactionalBatch batch = container.CreateTransactionalBatch(partitionKey);

Затем добавьте в этот пакет несколько операций:

Product bike = new (
    id: "68719520766",
    category: "road-bikes",
    name: "Chropen Road Bike"
);

batch.CreateItem<Product>(bike);

Part part = new (
    id: "68719519885",
    category: "road-bikes",
    name: "Tronosuros Tire",
    productId: bike.id
);

batch.CreateItem<Part>(part);

Наконец, вызовите для пакета ExecuteAsync:

using TransactionalBatchResponse response = await batch.ExecuteAsync();

Проверьте успешность выполнения в полученном ответе. Если ответ указывает на успешное выполнение, извлеките результаты:

if (response.IsSuccessStatusCode)
{
    TransactionalBatchOperationResult<Product> productResponse;
    productResponse = response.GetOperationResultAtIndex<Product>(0);
    Product productResult = productResponse.Resource;

    TransactionalBatchOperationResult<Part> partResponse;
    partResponse = response.GetOperationResultAtIndex<Part>(1);
    Part partResult = partResponse.Resource;
}

При создании транзакционной пакетной операции вызовите CosmosBatch.createCosmosBatch:

PartitionKey partitionKey = new PartitionKey("road-bikes");

CosmosBatch batch = CosmosBatch.createCosmosBatch(partitionKey);

Затем добавьте в этот пакет несколько операций:

Product bike = new Product();
bike.setId("68719520766");
bike.setCategory("road-bikes");
bike.setName("Chropen Road Bike");

batch.createItemOperation(bike);

Part part = new Part();
part.setId("68719519885");
part.setCategory("road-bikes");
part.setName("Tronosuros Tire");
part.setProductId(bike.getId());

batch.createItemOperation(part);

Наконец, вызовите для экземпляра контейнера операцию executeCosmosBatch с пакетом:

CosmosBatchResponse response = container.executeCosmosBatch(batch);

Проверьте успешность выполнения в полученном ответе. Если ответ указывает на успех, извлеките результаты.

if (response.isSuccessStatusCode())
{
    List<CosmosBatchOperationResult> results = response.getResults();
}

Получение или создание экземпляра контейнера:

container = database.create_container_if_not_exists(id="batch_container",
                                                        partition_key=PartitionKey(path='/category'))

В Python операции пакетной службы транзакций выглядят очень похожими на api сингулярных операций и содержат кортежи (operation_type_string, args_tuple, batch_operation_kwargs_dictionary). Ниже приведены примеры элементов, которые будут использоваться для демонстрации функций пакетных операций:

create_demo_item = {
    "id": "68719520766",
    "category": "road-bikes",
    "name": "Chropen Road Bike"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item1 = {
    "id": "68719519884",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item2 = {
    "id": "68719519886",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item3 = {
    "id": "68719519887",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, we'll upsert the item with id 68719519885
upsert_demo_item = {
    "id": "68719519885",
    "category": "road-bikes",
    "name": "Tronosuros Tire Upserted",
    "productId": "68719520768"
}

# for replace demo, we'll replace the read_demo_item2 with this item
replace_demo_item = {
    "id": "68719519886",
    "category": "road-bikes",
    "name": "Tronosuros Tire replaced",
    "productId": "68719520769"
}

# for replace with etag match demo, we'll replace the read_demo_item3 with this item
# The use of etags and if-match/if-none-match options allows users to run conditional replace operations
# based on the etag value passed. When using if-match, the request will only succeed if the item's latest etag
# matches the passed in value. For more on optimistic concurrency control, see the link below:
# https://learn.microsoft.com/azure/cosmos-db/nosql/database-transactions-optimistic-concurrency
replace_demo_item_if_match_operation = {
    "id": "68719519887",
    "category": "road-bikes",
    "name": "Tronosuros Tireh",
    "wasReplaced": "Replaced based on etag match"
    "productId": "68719520769"
}

Подготовьте операции для добавления в пакет:

create_item_operation = ("create", (create_demo_item,), {})
read_item_operation = ("read", ("68719519884",), {})
delete_item_operation = ("delete", ("68719519885",), {})
upsert_item_operation = ("upsert", (upsert_demo_item,), {})
replace_item_operation = ("replace", ("68719519886", replace_demo_item), {})
replace_item_if_match_operation = ("replace",
                                       ("68719519887", replace_demo_item_if_match_operation),
                                       {"if_match_etag": container.client_connection.last_response_headers.get("etag")})

Добавьте операции в пакет:

batch_operations = [
        create_item_operation,
        read_item_operation,
        delete_item_operation,
        upsert_item_operation,
        replace_item_operation,
        replace_item_if_match_operation
    ]

Наконец, выполните пакет:

try:
    # Run that list of operations
    batch_results = container.execute_item_batch(batch_operations=batch_operations, partition_key="road_bikes")
    # Batch results are returned as a list of item operation results - or raise a CosmosBatchOperationError if
    # one of the operations failed within your batch request.
    print("\nResults for the batch operations: {}\n".format(batch_results))
except exceptions.CosmosBatchOperationError as e:
        error_operation_index = e.error_index
        error_operation_response = e.operation_responses[error_operation_index]
        error_operation = batch_operations[error_operation_index]
        print("\nError operation: {}, error operation response: {}\n".format(error_operation, error_operation_response))
    # [END handle_batch_error]

Обратите внимание на использование операции обновления и операции replace_if_match_etag в составе
Словарь kwargs для пакетной операции ограничен и принимает только три разных ключевых значения. Если требуется использовать условное патчирование в пакете, доступно использование ключа filter_predicate для операции патча, а для использования etags с любой из операций доступны ключи if_match_etag/if_none_match_etag.

batch_operations = [
       ("replace", (item_id, item_body), {"if_match_etag": etag}),
       ("patch", (item_id, operations), {"filter_predicate": filter_predicate, "if_none_match_etag": etag}),
   ]

В случае сбоя проблемная операция возвращает код состояния с обозначением ошибки. Все остальные операции получат код состояния 424 (сбой зависимости). Если операция завершается ошибкой, так как она пытается создать уже существующий элемент, возвращается код состояния 409 (конфликт). Код состояния позволяет понять причину сбоя всей транзакции.

Как выполняются транзакционные пакетные операции

При выполнении пакетной службы транзакций все операции в пакете транзакций группируются, сериализуются в одну полезные данные и отправляются в виде одного запроса в службу Azure Cosmos DB.

Эта служба получает запрос и выполняет все операции в области транзакций, а затем возвращает ответ с помощью того же протокола сериализации. Этот ответ обозначает успех или сбой транзакции и предоставляет ответы по отдельным операциям.

Пакет SDK передает вам этот ответ для проверки результата. При желании вы можете извлечь каждый из результатов внутренних операций.

Ограничения

В настоящее время существует два известных ограничения:

• Ограничение размера запроса Azure Cosmos DB ограничивает размер полезной нагрузки транзакционного пакета до 2 МБ, и максимальное время выполнения составляет 5 секунд.
• Существует настоящее ограничение в 100 операций на транзакционный пакет, чтобы обеспечить производительность в соответствии с ожиданиями и соответствовать соглашениям об уровне обслуживания.

Следующий шаг