or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

database-accounts.mdindex.mdmongodb-resources.mdmonitoring-metrics.mdmulti-api-resources.mdnetworking-security.mdrestore-operations.mdservice-management.mdsql-resources.md
tile.json

tessl/pypi-azure-mgmt-cosmosdb

Microsoft Azure Cosmos DB Management Client Library for Python

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/azure-mgmt-cosmosdb@9.8.x

To install, run

npx @tessl/cli install tessl/pypi-azure-mgmt-cosmosdb@9.8.0

index.mddocs/

Azure Cosmos DB Management Client

A comprehensive Python client library for managing Azure Cosmos DB resources through the Azure Resource Manager APIs. This library enables developers to programmatically create, configure, and manage Cosmos DB accounts, databases, containers, and other related resources in Azure across all supported APIs (SQL, MongoDB, Cassandra, Gremlin, and Table).

Package Information

  • Package Name: azure-mgmt-cosmosdb
  • Package Type: pypi
  • Language: Python
  • Version: 9.8.0
  • Installation: pip install azure-mgmt-cosmosdb

Core Imports

from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.identity import DefaultAzureCredential

For async operations:

from azure.mgmt.cosmosdb.aio import CosmosDBManagementClient
from azure.identity.aio import DefaultAzureCredential

Basic Usage

from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.identity import DefaultAzureCredential

# Initialize client
credential = DefaultAzureCredential()
subscription_id = "your-subscription-id"
client = CosmosDBManagementClient(credential, subscription_id)

# List all Cosmos DB accounts in subscription
for account in client.database_accounts.list():
    print(f"Account: {account.name}, Location: {account.location}")

# Get specific account
account = client.database_accounts.get("resource-group", "account-name")
print(f"Account Kind: {account.kind}, API: {account.capabilities}")

# Create SQL database (Long Running Operation)
database_params = {
    "resource": {"id": "database-name"},
    "options": {"throughput": 400}
}
poller = client.sql_resources.begin_create_update_sql_database(
    "resource-group", "account-name", "database-name", database_params
)
result = poller.result()  # Wait for completion

Architecture

The Azure Cosmos DB Management Client follows the Azure SDK design patterns:

  • Client: CosmosDBManagementClient serves as the main entry point, managing authentication and providing access to operation groups
  • Operations: Specialized operation classes for different resource types (database accounts, SQL resources, MongoDB resources, etc.)
  • Models: Complete data classes representing Azure resources, configurations, and responses
  • Long Running Operations (LRO): Async operations return LROPoller objects for monitoring progress
  • Pagination: List operations return ItemPaged iterators for automatic result pagination

This design provides comprehensive coverage of the Azure Cosmos DB management plane while maintaining consistency with the broader Azure SDK ecosystem.

Capabilities

Database Account Management

Core database account lifecycle operations including creation, configuration, scaling, failover management, and key rotation. These operations manage the top-level Cosmos DB account resource that hosts all databases and containers.

class DatabaseAccountsOperations:
    def get(self, resource_group_name: str, account_name: str) -> DatabaseAccountGetResults: ...
    def begin_create_or_update(self, resource_group_name: str, account_name: str, create_update_parameters: DatabaseAccountCreateUpdateParameters) -> LROPoller[DatabaseAccountGetResults]: ...
    def begin_delete(self, resource_group_name: str, account_name: str) -> LROPoller[None]: ...
    def list(self) -> ItemPaged[DatabaseAccountGetResults]: ...
    def list_by_resource_group(self, resource_group_name: str) -> ItemPaged[DatabaseAccountGetResults]: ...
    def list_keys(self, resource_group_name: str, account_name: str) -> DatabaseAccountListKeysResult: ...
    def begin_regenerate_key(self, resource_group_name: str, account_name: str, key_to_regenerate: DatabaseAccountRegenerateKeyParameters) -> LROPoller[None]: ...
    def begin_failover_priority_change(self, resource_group_name: str, account_name: str, failover_parameters: FailoverPolicies) -> LROPoller[None]: ...

Database Account Management

SQL API Resource Management

Comprehensive management of SQL (Core) API resources including databases, containers, stored procedures, triggers, user-defined functions, client encryption keys, and RBAC role definitions.

class SqlResourcesOperations:
    def list_sql_databases(self, resource_group_name: str, account_name: str) -> ItemPaged[SqlDatabaseGetResults]: ...
    def get_sql_database(self, resource_group_name: str, account_name: str, database_name: str) -> SqlDatabaseGetResults: ...
    def begin_create_update_sql_database(self, resource_group_name: str, account_name: str, database_name: str, create_update_sql_database_parameters: SqlDatabaseCreateUpdateParameters) -> LROPoller[SqlDatabaseGetResults]: ...
    def list_sql_containers(self, resource_group_name: str, account_name: str, database_name: str) -> ItemPaged[SqlContainerGetResults]: ...
    def get_sql_container(self, resource_group_name: str, account_name: str, database_name: str, container_name: str) -> SqlContainerGetResults: ...
    def begin_create_update_sql_container(self, resource_group_name: str, account_name: str, database_name: str, container_name: str, create_update_sql_container_parameters: SqlContainerCreateUpdateParameters) -> LROPoller[SqlContainerGetResults]: ...
    def get_sql_database_throughput(self, resource_group_name: str, account_name: str, database_name: str) -> ThroughputSettingsGetResults: ...
    def begin_update_sql_database_throughput(self, resource_group_name: str, account_name: str, database_name: str, update_throughput_parameters: ThroughputSettingsUpdateParameters) -> LROPoller[ThroughputSettingsGetResults]: ...

SQL API Resources

MongoDB API Resource Management

Full management of MongoDB API resources including databases, collections, role definitions, and user definitions with support for MongoDB-specific features like sharding and indexing.

class MongoDBResourcesOperations:
    def list_mongo_db_databases(self, resource_group_name: str, account_name: str) -> ItemPaged[MongoDBDatabaseGetResults]: ...
    def get_mongo_db_database(self, resource_group_name: str, account_name: str, database_name: str) -> MongoDBDatabaseGetResults: ...
    def begin_create_update_mongo_db_database(self, resource_group_name: str, account_name: str, database_name: str, create_update_mongo_db_database_parameters: MongoDBDatabaseCreateUpdateParameters) -> LROPoller[MongoDBDatabaseGetResults]: ...
    def list_mongo_db_collections(self, resource_group_name: str, account_name: str, database_name: str) -> ItemPaged[MongoDBCollectionGetResults]: ...
    def get_mongo_db_collection(self, resource_group_name: str, account_name: str, database_name: str, collection_name: str) -> MongoDBCollectionGetResults: ...
    def begin_create_update_mongo_db_collection(self, resource_group_name: str, account_name: str, database_name: str, collection_name: str, create_update_mongo_db_collection_parameters: MongoDBCollectionCreateUpdateParameters) -> LROPoller[MongoDBCollectionGetResults]: ...

MongoDB API Resources

Multi-API Resource Management

Management operations for Table API, Cassandra API, and Gremlin API resources, providing consistent CRUD operations and throughput management across all Cosmos DB APIs.

class TableResourcesOperations:
    def list_tables(self, resource_group_name: str, account_name: str) -> ItemPaged[TableGetResults]: ...
    def get_table(self, resource_group_name: str, account_name: str, table_name: str) -> TableGetResults: ...
    def begin_create_update_table(self, resource_group_name: str, account_name: str, table_name: str, create_update_table_parameters: TableCreateUpdateParameters) -> LROPoller[TableGetResults]: ...

class CassandraResourcesOperations:
    def list_cassandra_keyspaces(self, resource_group_name: str, account_name: str) -> ItemPaged[CassandraKeyspaceGetResults]: ...
    def list_cassandra_tables(self, resource_group_name: str, account_name: str, keyspace_name: str) -> ItemPaged[CassandraTableGetResults]: ...

class GremlinResourcesOperations:
    def list_gremlin_databases(self, resource_group_name: str, account_name: str) -> ItemPaged[GremlinDatabaseGetResults]: ...
    def list_gremlin_graphs(self, resource_group_name: str, account_name: str, database_name: str) -> ItemPaged[GremlinGraphGetResults]: ...

Multi-API Resources

Point-in-Time Restore Operations

Comprehensive point-in-time restore capabilities for discovering restorable resources across all APIs and time ranges, enabling precise backup and recovery scenarios.

class RestorableDatabaseAccountsOperations:
    def list(self) -> ItemPaged[RestorableDatabaseAccountGetResult]: ...
    def list_by_location(self, location: str) -> ItemPaged[RestorableDatabaseAccountGetResult]: ...

class RestorableSqlResourcesOperations:
    def list(self, location: str, instance_id: str) -> ItemPaged[RestorableSqlResourcesGetResult]: ...

class RestorableMongodbResourcesOperations:
    def list(self, location: str, instance_id: str) -> ItemPaged[RestorableMongodbResourcesGetResult]: ...

Point-in-Time Restore

Monitoring and Metrics

Comprehensive monitoring capabilities including percentile metrics, usage statistics, and performance monitoring across regions, databases, containers, and partitions.

class PercentileOperations:
    def list_metrics(self, resource_group_name: str, account_name: str, filter: str) -> ItemPaged[PercentileMetric]: ...

class DatabaseOperations:
    def list_metrics(self, resource_group_name: str, account_name: str, database_rid: str, filter: str) -> ItemPaged[Metric]: ...
    def list_usages(self, resource_group_name: str, account_name: str, database_rid: str, filter: str = None) -> ItemPaged[Usage]: ...

class CollectionOperations:
    def list_metrics(self, resource_group_name: str, account_name: str, database_rid: str, collection_rid: str, filter: str) -> ItemPaged[Metric]: ...

Monitoring and Metrics

Networking and Security

Private endpoint connections, private link resources, and security configuration for secure network access to Cosmos DB accounts.

class PrivateEndpointConnectionsOperations:
    def list_by_database_account(self, resource_group_name: str, account_name: str) -> ItemPaged[PrivateEndpointConnection]: ...
    def get(self, resource_group_name: str, account_name: str, private_endpoint_connection_name: str) -> PrivateEndpointConnection: ...
    def begin_create_or_update(self, resource_group_name: str, account_name: str, private_endpoint_connection_name: str, parameters: PrivateEndpointConnection) -> LROPoller[PrivateEndpointConnection]: ...

class PrivateLinkResourcesOperations:
    def list_by_database_account(self, resource_group_name: str, account_name: str) -> ItemPaged[PrivateLinkResource]: ...
    def get(self, resource_group_name: str, account_name: str, group_name: str) -> PrivateLinkResource: ...

Networking and Security

Provider Operations

Core Azure resource provider operations for listing available Cosmos DB management operations and discovering API capabilities.

class Operations:
    def list(self) -> ItemPaged[Operation]: ...

Location Management

Azure region and location operations for discovering available Cosmos DB regions and their capabilities.

class LocationsOperations:
    def list(self) -> ItemPaged[LocationGetResult]: ...

Service Management

Management of Cosmos DB services, Cassandra clusters, data centers, and notebook workspaces for advanced scenarios and specialized deployments.

class ServiceOperations:
    def list(self, resource_group_name: str, account_name: str) -> ItemPaged[ServiceResource]: ...
    def get(self, resource_group_name: str, account_name: str, service_name: str) -> ServiceResource: ...
    def begin_create(self, resource_group_name: str, account_name: str, service_name: str, create_service_parameters: ServiceResourceCreateUpdateParameters) -> LROPoller[ServiceResource]: ...

class CassandraClustersOperations:
    def list_by_subscription(self) -> ItemPaged[ClusterResource]: ...
    def list_by_resource_group(self, resource_group_name: str) -> ItemPaged[ClusterResource]: ...
    def get(self, resource_group_name: str, cluster_name: str) -> ClusterResource: ...
    def begin_create_update(self, resource_group_name: str, cluster_name: str, body: ClusterResource) -> LROPoller[ClusterResource]: ...

class NotebookWorkspacesOperations:
    def list_by_database_account(self, resource_group_name: str, account_name: str) -> ItemPaged[NotebookWorkspace]: ...
    def get(self, resource_group_name: str, account_name: str, notebook_workspace_name: str) -> NotebookWorkspace: ...
    def begin_create_or_update(self, resource_group_name: str, account_name: str, notebook_workspace_name: str, notebook_create_update_parameters: NotebookWorkspaceCreateUpdateParameters) -> LROPoller[NotebookWorkspace]: ...

Service Management

Core Types

class CosmosDBManagementClient:
    def __init__(
        self,
        credential: TokenCredential,
        subscription_id: str,
        base_url: Optional[str] = None,
        **kwargs: Any
    ) -> None: ...
    
    def close(self) -> None: ...
    def __enter__(self) -> Self: ...
    def __exit__(self, *exc_details: Any) -> None: ...

# Long Running Operation Poller
class LROPoller[T]:
    def result(self, timeout: Optional[float] = None) -> T: ...
    def status(self) -> str: ...
    def done(self) -> bool: ...
    def wait(self, timeout: Optional[float] = None) -> None: ...

# Pagination Iterator
class ItemPaged[T]:
    def __iter__(self) -> Iterator[T]: ...
    def by_page(self, continuation_token: Optional[str] = None) -> Iterator[Iterator[T]]: ...