Microsoft Azure Cosmos DB Management Client Library for Python
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Full management of MongoDB API resources including databases, collections, role definitions, and user definitions. The MongoDB API provides compatibility with MongoDB drivers and tools while leveraging Cosmos DB's global distribution and enterprise features.
Create, configure, and manage MongoDB databases with throughput and scaling options.
def list_mongo_db_databases(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[MongoDBDatabaseGetResults]:
"""
List MongoDB databases in a Cosmos DB account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
ItemPaged[MongoDBDatabaseGetResults]: Paginated list of MongoDB databases
"""
def get_mongo_db_database(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> MongoDBDatabaseGetResults:
"""
Get properties of a MongoDB database.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
Returns:
MongoDBDatabaseGetResults: MongoDB database properties and configuration
"""
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]:
"""
Create or update a MongoDB database (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database to create/update
- create_update_mongo_db_database_parameters: Database configuration parameters
Returns:
LROPoller[MongoDBDatabaseGetResults]: Poller for monitoring operation progress
"""
def begin_delete_mongo_db_database(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> LROPoller[None]:
"""
Delete a MongoDB database (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Manage MongoDB collections with sharding, indexing, and MongoDB-specific features.
def list_mongo_db_collections(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> ItemPaged[MongoDBCollectionGetResults]:
"""
List MongoDB collections in a database.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
Returns:
ItemPaged[MongoDBCollectionGetResults]: Paginated list of MongoDB collections
"""
def get_mongo_db_collection(
self,
resource_group_name: str,
account_name: str,
database_name: str,
collection_name: str
) -> MongoDBCollectionGetResults:
"""
Get properties of a MongoDB collection.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
- collection_name: Name of the MongoDB collection
Returns:
MongoDBCollectionGetResults: MongoDB collection properties and configuration
"""
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]:
"""
Create or update a MongoDB collection (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
- collection_name: Name of the MongoDB collection to create/update
- create_update_mongo_db_collection_parameters: Collection configuration parameters
Returns:
LROPoller[MongoDBCollectionGetResults]: Poller for monitoring operation progress
"""
def begin_delete_mongo_db_collection(
self,
resource_group_name: str,
account_name: str,
database_name: str,
collection_name: str
) -> LROPoller[None]:
"""
Delete a MongoDB collection (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
- collection_name: Name of the MongoDB collection to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Manage throughput settings for databases and collections with autoscale and manual options.
def get_mongo_db_database_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> ThroughputSettingsGetResults:
"""
Get throughput settings for a MongoDB database.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
Returns:
ThroughputSettingsGetResults: Current throughput configuration
"""
def begin_update_mongo_db_database_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str,
update_throughput_parameters: ThroughputSettingsUpdateParameters
) -> LROPoller[ThroughputSettingsGetResults]:
"""
Update throughput settings for a MongoDB database (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
- update_throughput_parameters: New throughput configuration
Returns:
LROPoller[ThroughputSettingsGetResults]: Poller for monitoring operation progress
"""
def get_mongo_db_collection_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str,
collection_name: str
) -> ThroughputSettingsGetResults:
"""
Get throughput settings for a MongoDB collection.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
- collection_name: Name of the MongoDB collection
Returns:
ThroughputSettingsGetResults: Current throughput configuration
"""
def begin_migrate_mongo_db_database_to_autoscale(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> LROPoller[ThroughputSettingsGetResults]:
"""
Migrate MongoDB database from manual to autoscale throughput (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
Returns:
LROPoller[ThroughputSettingsGetResults]: Poller for monitoring operation progress
"""
def begin_migrate_mongo_db_collection_to_manual_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str,
collection_name: str
) -> LROPoller[ThroughputSettingsGetResults]:
"""
Migrate MongoDB collection from autoscale to manual throughput (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the MongoDB database
- collection_name: Name of the MongoDB collection
Returns:
LROPoller[ThroughputSettingsGetResults]: Poller for monitoring operation progress
"""Manage custom role definitions for MongoDB API security and access control.
def list_mongo_db_role_definitions(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[MongoRoleDefinitionGetResults]:
"""
List MongoDB role definitions for access control.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
ItemPaged[MongoRoleDefinitionGetResults]: Paginated list of role definitions
"""
def get_mongo_db_role_definition(
self,
resource_group_name: str,
account_name: str,
mongo_role_definition_id: str
) -> MongoRoleDefinitionGetResults:
"""
Get properties of a MongoDB role definition.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- mongo_role_definition_id: Unique identifier of the role definition
Returns:
MongoRoleDefinitionGetResults: Role definition properties and permissions
"""
def begin_create_update_mongo_db_role_definition(
self,
resource_group_name: str,
account_name: str,
mongo_role_definition_id: str,
create_update_mongo_role_definition_parameters: MongoRoleDefinitionCreateUpdateParameters
) -> LROPoller[MongoRoleDefinitionGetResults]:
"""
Create or update a MongoDB role definition (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- mongo_role_definition_id: Unique identifier of the role definition
- create_update_mongo_role_definition_parameters: Role definition configuration
Returns:
LROPoller[MongoRoleDefinitionGetResults]: Poller for monitoring operation progress
"""
def begin_delete_mongo_db_role_definition(
self,
resource_group_name: str,
account_name: str,
mongo_role_definition_id: str
) -> LROPoller[None]:
"""
Delete a MongoDB role definition (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- mongo_role_definition_id: Unique identifier of the role definition
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Manage user definitions with role assignments for MongoDB API authentication.
def list_mongo_db_user_definitions(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[MongoUserDefinitionGetResults]:
"""
List MongoDB user definitions.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
ItemPaged[MongoUserDefinitionGetResults]: Paginated list of user definitions
"""
def get_mongo_db_user_definition(
self,
resource_group_name: str,
account_name: str,
mongo_user_definition_id: str
) -> MongoUserDefinitionGetResults:
"""
Get properties of a MongoDB user definition.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- mongo_user_definition_id: Unique identifier of the user definition
Returns:
MongoUserDefinitionGetResults: User definition properties and roles
"""
def begin_create_update_mongo_db_user_definition(
self,
resource_group_name: str,
account_name: str,
mongo_user_definition_id: str,
create_update_mongo_user_definition_parameters: MongoUserDefinitionCreateUpdateParameters
) -> LROPoller[MongoUserDefinitionGetResults]:
"""
Create or update a MongoDB user definition (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- mongo_user_definition_id: Unique identifier of the user definition
- create_update_mongo_user_definition_parameters: User definition configuration
Returns:
LROPoller[MongoUserDefinitionGetResults]: Poller for monitoring operation progress
"""
def begin_delete_mongo_db_user_definition(
self,
resource_group_name: str,
account_name: str,
mongo_user_definition_id: str
) -> LROPoller[None]:
"""
Delete a MongoDB user definition (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- mongo_user_definition_id: Unique identifier of the user definition
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.mgmt.cosmosdb.models import (
MongoDBDatabaseCreateUpdateParameters,
MongoDBDatabaseResource,
MongoDBCollectionCreateUpdateParameters,
MongoDBCollectionResource,
MongoIndex,
MongoIndexKeys,
CreateUpdateOptions
)
from azure.identity import DefaultAzureCredential
client = CosmosDBManagementClient(DefaultAzureCredential(), "subscription-id")
# Create MongoDB database with shared throughput
database_params = MongoDBDatabaseCreateUpdateParameters(
resource=MongoDBDatabaseResource(id="inventory"),
options=CreateUpdateOptions(throughput=800)
)
poller = client.mongo_db_resources.begin_create_update_mongo_db_database(
"my-resource-group",
"my-cosmos-account",
"inventory",
database_params
)
database = poller.result()
# Create MongoDB collection with sharding and indexes
collection_resource = MongoDBCollectionResource(
id="products",
shard_key={"category": "Hash"}, # Shard key for partitioning
indexes=[
MongoIndex(
key=MongoIndexKeys(keys=["name", "category"]),
options={"unique": False, "background": True}
),
MongoIndex(
key=MongoIndexKeys(keys=["price"]),
options={"name": "price_index"}
)
]
)
collection_params = MongoDBCollectionCreateUpdateParameters(
resource=collection_resource,
options=CreateUpdateOptions(throughput=400) # Dedicated throughput
)
poller = client.mongo_db_resources.begin_create_update_mongo_db_collection(
"my-resource-group",
"my-cosmos-account",
"inventory",
"products",
collection_params
)
collection = poller.result()
print(f"Created collection: {collection.name} with shard key: {collection.resource.shard_key}")from azure.mgmt.cosmosdb.models import (
MongoRoleDefinitionCreateUpdateParameters,
MongoRoleDefinitionResource,
Role,
Privilege,
MongoUserDefinitionCreateUpdateParameters,
MongoUserDefinitionResource
)
# Create custom role definition
role_resource = MongoRoleDefinitionResource(
role_name="ProductManager",
type="CustomRole",
database_name="inventory",
roles=[], # Inherited roles
privileges=[
Privilege(
resource={"db": "inventory", "collection": "products"},
actions=["insert", "update", "remove", "find"]
)
]
)
role_params = MongoRoleDefinitionCreateUpdateParameters(resource=role_resource)
poller = client.mongo_db_resources.begin_create_update_mongo_db_role_definition(
"my-resource-group",
"my-cosmos-account",
"product-manager-role",
role_params
)
role_def = poller.result()
# Create user with role assignment
user_resource = MongoUserDefinitionResource(
user_name="product_admin",
password="secure_password_123",
database_name="inventory",
custom_data="Product management user",
roles=[
Role(
role="ProductManager",
db="inventory"
)
],
mechanisms="SCRAM-SHA-256"
)
user_params = MongoUserDefinitionCreateUpdateParameters(resource=user_resource)
poller = client.mongo_db_resources.begin_create_update_mongo_db_user_definition(
"my-resource-group",
"my-cosmos-account",
"product-admin-user",
user_params
)
user_def = poller.result()
print(f"Created user: {user_def.resource.user_name} with roles: {[r.role for r in user_def.resource.roles]}")class MongoDBDatabaseGetResults:
"""MongoDB database properties and configuration."""
id: str
name: str
type: str
resource: MongoDBDatabaseGetPropertiesResource
options: MongoDBDatabaseGetPropertiesOptions
class MongoDBDatabaseResource:
"""MongoDB database resource definition."""
id: str # Database name
class MongoDBCollectionGetResults:
"""MongoDB collection properties and configuration."""
id: str
name: str
type: str
resource: MongoDBCollectionGetPropertiesResource
options: MongoDBCollectionGetPropertiesOptions
class MongoDBCollectionResource:
"""MongoDB collection resource definition."""
id: str # Collection name
shard_key: Dict[str, str] # Shard key for partitioning (e.g., {"category": "Hash"})
indexes: List[MongoIndex] # Collection indexes
analytical_storage_ttl: int # TTL for analytical storage
class MongoIndex:
"""MongoDB index definition."""
key: MongoIndexKeys # Index key specification
options: Dict[str, Any] # Index options (unique, background, etc.)
class MongoIndexKeys:
"""MongoDB index key specification."""
keys: List[str] # Fields to index
class MongoRoleDefinitionGetResults:
"""MongoDB role definition for access control."""
id: str
name: str
type: str
resource: MongoRoleDefinitionResource
class MongoRoleDefinitionResource:
"""MongoDB role definition resource."""
role_name: str # Name of the role
type: str # "BuiltInRole" or "CustomRole"
database_name: str # Database where role is defined
roles: List[Role] # Inherited roles
privileges: List[Privilege] # Role privileges
class Privilege:
"""MongoDB role privilege definition."""
resource: Dict[str, str] # Resource specification (db, collection)
actions: List[str] # Allowed actions (insert, update, remove, find, etc.)
class MongoUserDefinitionGetResults:
"""MongoDB user definition properties."""
id: str
name: str
type: str
resource: MongoUserDefinitionResource
class MongoUserDefinitionResource:
"""MongoDB user definition resource."""
user_name: str # Username
password: str # User password
database_name: str # Authentication database
custom_data: str # Custom user data
roles: List[Role] # Assigned roles
mechanisms: str # Authentication mechanism ("SCRAM-SHA-1", "SCRAM-SHA-256")
class Role:
"""MongoDB role assignment."""
role: str # Role name
db: str # Database where role appliesInstall with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-cosmosdb