tessl/pypi-azure-mgmt-cosmosdb
Microsoft Azure Cosmos DB Management Client Library for Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
service-management.mddocs/
Service Management
Management of Cosmos DB services, Cassandra clusters, data centers, and notebook workspaces for advanced scenarios and specialized deployments. These operations enable management of additional Cosmos DB features and managed services.
Capabilities
Cosmos DB Service Management
Manage additional services and features associated with Cosmos DB accounts.
def list(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[ServiceResource]:
"""
List services associated with a Cosmos DB account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
ItemPaged[ServiceResource]: Paginated list of services
"""
def get(
self,
resource_group_name: str,
account_name: str,
service_name: str
) -> ServiceResource:
"""
Get properties of a Cosmos DB service.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- service_name: Name of the service
Returns:
ServiceResource: Service properties and configuration
"""
def begin_create(
self,
resource_group_name: str,
account_name: str,
service_name: str,
create_service_parameters: ServiceResourceCreateUpdateParameters
) -> LROPoller[ServiceResource]:
"""
Create a new service for a Cosmos DB account (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- service_name: Name of the service to create
- create_service_parameters: Service configuration parameters
Returns:
LROPoller[ServiceResource]: Poller for monitoring operation progress
"""
def begin_update(
self,
resource_group_name: str,
account_name: str,
service_name: str,
update_service_parameters: ServiceResourceCreateUpdateParameters
) -> LROPoller[ServiceResource]:
"""
Update a service configuration (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- service_name: Name of the service to update
- update_service_parameters: Updated service configuration
Returns:
LROPoller[ServiceResource]: Poller for monitoring operation progress
"""
def begin_delete(
self,
resource_group_name: str,
account_name: str,
service_name: str
) -> LROPoller[None]:
"""
Delete a service (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- service_name: Name of the service to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Managed Cassandra Cluster Management
Manage Azure Managed Instance for Apache Cassandra clusters with full lifecycle operations.
def list_by_subscription(self) -> ItemPaged[ClusterResource]:
"""
List all Cassandra clusters in the subscription.
Returns:
ItemPaged[ClusterResource]: Paginated list of all Cassandra clusters
"""
def list_by_resource_group(
self,
resource_group_name: str
) -> ItemPaged[ClusterResource]:
"""
List Cassandra clusters in a resource group.
Parameters:
- resource_group_name: Name of the resource group
Returns:
ItemPaged[ClusterResource]: Paginated list of clusters in the resource group
"""
def get(
self,
resource_group_name: str,
cluster_name: str
) -> ClusterResource:
"""
Get properties of a Cassandra cluster.
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the Cassandra cluster
Returns:
ClusterResource: Cluster properties, status, and configuration
"""
def begin_create_update(
self,
resource_group_name: str,
cluster_name: str,
body: ClusterResource
) -> LROPoller[ClusterResource]:
"""
Create or update a Cassandra cluster (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the cluster to create/update
- body: Cluster configuration and properties
Returns:
LROPoller[ClusterResource]: Poller for monitoring operation progress
"""
def begin_delete(
self,
resource_group_name: str,
cluster_name: str
) -> LROPoller[None]:
"""
Delete a Cassandra cluster (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the cluster to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""
def begin_update(
self,
resource_group_name: str,
cluster_name: str,
body: ClusterResource
) -> LROPoller[ClusterResource]:
"""
Update a Cassandra cluster configuration (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the cluster to update
- body: Updated cluster configuration
Returns:
LROPoller[ClusterResource]: Poller for monitoring operation progress
"""
def begin_invoke_command(
self,
resource_group_name: str,
cluster_name: str,
body: CommandPostBody
) -> LROPoller[CommandOutput]:
"""
Invoke a command on the Cassandra cluster (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the Cassandra cluster
- body: Command to execute and parameters
Returns:
LROPoller[CommandOutput]: Poller for monitoring command execution
"""Cassandra Data Center Management
Manage data centers within Cassandra clusters for multi-region deployments.
def list(
self,
resource_group_name: str,
cluster_name: str
) -> ItemPaged[DataCenterResource]:
"""
List data centers in a Cassandra cluster.
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the Cassandra cluster
Returns:
ItemPaged[DataCenterResource]: Paginated list of data centers
"""
def get(
self,
resource_group_name: str,
cluster_name: str,
data_center_name: str
) -> DataCenterResource:
"""
Get properties of a Cassandra data center.
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the Cassandra cluster
- data_center_name: Name of the data center
Returns:
DataCenterResource: Data center properties and configuration
"""
def begin_create_update(
self,
resource_group_name: str,
cluster_name: str,
data_center_name: str,
body: DataCenterResource
) -> LROPoller[DataCenterResource]:
"""
Create or update a data center (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the Cassandra cluster
- data_center_name: Name of the data center to create/update
- body: Data center configuration
Returns:
LROPoller[DataCenterResource]: Poller for monitoring operation progress
"""
def begin_delete(
self,
resource_group_name: str,
cluster_name: str,
data_center_name: str
) -> LROPoller[None]:
"""
Delete a data center (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the Cassandra cluster
- data_center_name: Name of the data center to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""
def begin_update(
self,
resource_group_name: str,
cluster_name: str,
data_center_name: str,
body: DataCenterResource
) -> LROPoller[DataCenterResource]:
"""
Update a data center configuration (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- cluster_name: Name of the Cassandra cluster
- data_center_name: Name of the data center to update
- body: Updated data center configuration
Returns:
LROPoller[DataCenterResource]: Poller for monitoring operation progress
"""Notebook Workspace Management
Manage Jupyter notebook workspaces for data exploration and analytics with Cosmos DB.
def list_by_database_account(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[NotebookWorkspace]:
"""
List notebook workspaces for a Cosmos DB account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
ItemPaged[NotebookWorkspace]: Paginated list of notebook workspaces
"""
def get(
self,
resource_group_name: str,
account_name: str,
notebook_workspace_name: str
) -> NotebookWorkspace:
"""
Get properties of a notebook workspace.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- notebook_workspace_name: Name of the notebook workspace
Returns:
NotebookWorkspace: Notebook workspace properties and status
"""
def begin_create_or_update(
self,
resource_group_name: str,
account_name: str,
notebook_workspace_name: str,
notebook_create_update_parameters: NotebookWorkspaceCreateUpdateParameters
) -> LROPoller[NotebookWorkspace]:
"""
Create or update a notebook workspace (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- notebook_workspace_name: Name of the workspace to create/update
- notebook_create_update_parameters: Workspace configuration parameters
Returns:
LROPoller[NotebookWorkspace]: Poller for monitoring operation progress
"""
def begin_delete(
self,
resource_group_name: str,
account_name: str,
notebook_workspace_name: str
) -> LROPoller[None]:
"""
Delete a notebook workspace (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- notebook_workspace_name: Name of the workspace to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""
def begin_start(
self,
resource_group_name: str,
account_name: str,
notebook_workspace_name: str
) -> LROPoller[None]:
"""
Start a notebook workspace (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- notebook_workspace_name: Name of the workspace to start
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""
def list_connection_info(
self,
resource_group_name: str,
account_name: str,
notebook_workspace_name: str
) -> NotebookWorkspaceConnectionInfoResult:
"""
Get connection information for a notebook workspace.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- notebook_workspace_name: Name of the notebook workspace
Returns:
NotebookWorkspaceConnectionInfoResult: Connection URLs and authentication tokens
"""
def regenerate_auth_token(
self,
resource_group_name: str,
account_name: str,
notebook_workspace_name: str
) -> None:
"""
Regenerate authentication token for a notebook workspace.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- notebook_workspace_name: Name of the notebook workspace
"""Usage Examples
Creating a Managed Cassandra Cluster
from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.mgmt.cosmosdb.models import (
ClusterResource,
ClusterResourceProperties,
AuthenticationMethod,
CassandraAuditLoggingPolicy,
SeedNode
)
from azure.identity import DefaultAzureCredential
client = CosmosDBManagementClient(DefaultAzureCredential(), "subscription-id")
# Create managed Cassandra cluster
cluster_config = ClusterResource(
location="East US",
properties=ClusterResourceProperties(
delegated_management_subnet_id="/subscriptions/sub-id/resourceGroups/network-rg/providers/Microsoft.Network/virtualNetworks/vnet/subnets/cassandra-subnet",
cassandra_version="3.11",
cluster_name_override="my-cassandra-cluster",
authentication_method=AuthenticationMethod.CASSANDRA,
initial_cassandra_admin_password="SecurePassword123!",
hourly_based_backup_options={
"retention_period_in_hours": 168 # 7 days
},
audit_logging_enabled=True,
repair_enabled=True,
client_certificates=[]
)
)
poller = client.cassandra_clusters.begin_create_update(
"my-resource-group",
"my-cassandra-cluster",
cluster_config
)
cluster = poller.result()
print(f"Created cluster: {cluster.name}")
print(f"Cluster status: {cluster.properties.provisioning_state}")
print(f"Seed nodes: {cluster.properties.seed_nodes}")Adding Data Centers to Cassandra Cluster
from azure.mgmt.cosmosdb.models import DataCenterResource, DataCenterResourceProperties
# Add data center in West US
datacenter_config = DataCenterResource(
properties=DataCenterResourceProperties(
data_center_location="West US",
delegated_management_subnet_id="/subscriptions/sub-id/resourceGroups/network-rg-west/providers/Microsoft.Network/virtualNetworks/vnet-west/subnets/cassandra-subnet",
node_count=3,
sku="Standard_DS14_v2",
disk_sku="P50",
availability_zone=True,
backup_storage_customer_key_uri=None,
managed_disk_customer_key_uri=None
)
)
poller = client.cassandra_data_centers.begin_create_update(
"my-resource-group",
"my-cassandra-cluster",
"dc-west",
datacenter_config
)
datacenter = poller.result()
print(f"Created data center: {datacenter.name}")
print(f"Node count: {datacenter.properties.node_count}")
print(f"Data center location: {datacenter.properties.data_center_location}")Setting Up Notebook Workspace
from azure.mgmt.cosmosdb.models import NotebookWorkspaceCreateUpdateParameters
# Create notebook workspace
notebook_config = NotebookWorkspaceCreateUpdateParameters()
poller = client.notebook_workspaces.begin_create_or_update(
"my-resource-group",
"my-cosmos-account",
"default", # Standard workspace name
notebook_config
)
workspace = poller.result()
print(f"Created notebook workspace: {workspace.name}")
# Start the workspace
start_poller = client.notebook_workspaces.begin_start(
"my-resource-group",
"my-cosmos-account",
"default"
)
start_poller.result()
# Get connection information
connection_info = client.notebook_workspaces.list_connection_info(
"my-resource-group",
"my-cosmos-account",
"default"
)
print(f"Notebook server endpoint: {connection_info.notebook_server_endpoint}")
print(f"Auth token: {connection_info.auth_token}")Managing Cosmos DB Services
from azure.mgmt.cosmosdb.models import (
ServiceResourceCreateUpdateParameters,
DataTransferServiceResourceProperties
)
# Create a data transfer service
service_config = ServiceResourceCreateUpdateParameters(
properties=DataTransferServiceResourceProperties(
service_type="DataTransfer",
instance_count=1,
instance_size="Cosmos.D4s"
)
)
poller = client.service.begin_create(
"my-resource-group",
"my-cosmos-account",
"DataTransfer",
service_config
)
service = poller.result()
print(f"Created service: {service.name}")
print(f"Service type: {service.properties.service_type}")
print(f"Status: {service.properties.status}")
# List all services
services = client.service.list("my-resource-group", "my-cosmos-account")
for svc in services:
print(f"Service: {svc.name}, Type: {svc.properties.service_type}, Status: {svc.properties.status}")Executing Commands on Cassandra Cluster
from azure.mgmt.cosmosdb.models import CommandPostBody
# Execute nodetool status command
command_body = CommandPostBody(
command="nodetool status",
arguments={},
host="10.0.0.4" # Specific node IP
)
poller = client.cassandra_clusters.begin_invoke_command(
"my-resource-group",
"my-cassandra-cluster",
command_body
)
command_result = poller.result()
print(f"Command output: {command_result.result}")Key Types
class ServiceResource:
"""Cosmos DB service resource configuration and status."""
id: str # Resource ID
name: str # Service name
type: str # Resource type
properties: ServiceResourceProperties # Service-specific properties
class ServiceResourceProperties:
"""Base properties for Cosmos DB services."""
service_type: str # Type of service ("DataTransfer", "SqlDedicatedGateway", etc.)
instance_size: str # Instance size (e.g., "Cosmos.D4s", "Cosmos.D8s")
instance_count: int # Number of instances
status: str # Service status ("Creating", "Running", "Updating", "Deleting", "Error")
creation_time: str # Service creation timestamp
class ClusterResource:
"""Managed Cassandra cluster configuration and status."""
id: str # Resource ID
name: str # Cluster name
type: str # Resource type
location: str # Azure region
tags: Dict[str, str] # Resource tags
properties: ClusterResourceProperties # Cluster properties
class ClusterResourceProperties:
"""Managed Cassandra cluster properties."""
provisioning_state: str # Provisioning status
cluster_name_override: str # Custom cluster name
authentication_method: str # Authentication method ("None", "Cassandra", "Ldap")
initial_cassandra_admin_password: str # Initial admin password
client_certificates: List[Certificate] # Client certificates for authentication
external_gossip_certificates: List[Certificate] # Gossip certificates
external_seed_nodes: List[SeedNode] # External seed nodes
seed_nodes: List[SeedNode] # Current seed nodes
hours_between_backups: int # Backup frequency in hours
cassandra_version: str # Cassandra version
repair_enabled: bool # Whether automatic repair is enabled
deallocated: bool # Whether cluster is deallocated
class DataCenterResource:
"""Cassandra data center configuration and status."""
id: str # Resource ID
name: str # Data center name
type: str # Resource type
properties: DataCenterResourceProperties # Data center properties
class DataCenterResourceProperties:
"""Cassandra data center properties."""
provisioning_state: str # Provisioning status
data_center_location: str # Azure region for the data center
delegated_management_subnet_id: str # Subnet ID for management traffic
node_count: int # Number of Cassandra nodes
seed_nodes: List[SeedNode] # Seed nodes in this data center
base64_encoded_cassandra_yaml_fragment: str # Cassandra.yaml configuration fragment
managed_disk_customer_key_uri: str # Customer-managed key for disk encryption
backup_storage_customer_key_uri: str # Customer-managed key for backup encryption
sku: str # VM SKU for nodes
disk_sku: str # Managed disk SKU
disk_capacity: int # Disk capacity in GB
availability_zone: bool # Whether to use availability zones
class NotebookWorkspace:
"""Notebook workspace configuration and status."""
id: str # Resource ID
name: str # Workspace name
type: str # Resource type
status: str # Workspace status ("Creating", "Online", "Offline", "Updating", "Deleting")
class NotebookWorkspaceConnectionInfoResult:
"""Notebook workspace connection information."""
auth_token: str # Authentication token for accessing the workspace
notebook_server_endpoint: str # Jupyter notebook server URL
class SeedNode:
"""Cassandra seed node information."""
ip_address: str # IP address of the seed node
class Certificate:
"""Certificate for authentication."""
pem: str # PEM-encoded certificate content
class CommandPostBody:
"""Command execution request for Cassandra clusters."""
command: str # Command to execute (e.g., "nodetool status")
arguments: Dict[str, str] # Command arguments
host: str # Target host IP address
class CommandOutput:
"""Result of command execution on Cassandra cluster."""
result: str # Command output
exit_code: int # Command exit codeAdvanced Management Scenarios
Multi-Region Cassandra Deployment
def deploy_multi_region_cassandra():
"""Deploy a Cassandra cluster across multiple regions."""
# Create primary cluster in East US
primary_cluster = create_cassandra_cluster("East US", "primary-dc")
# Wait for primary cluster to be ready
while primary_cluster.properties.provisioning_state != "Succeeded":
time.sleep(30)
primary_cluster = client.cassandra_clusters.get("my-rg", "my-cluster")
# Add data centers in other regions
regions = ["West US", "North Europe", "Southeast Asia"]
for i, region in enumerate(regions):
dc_name = f"dc-{i+2}"
datacenter_config = DataCenterResource(
properties=DataCenterResourceProperties(
data_center_location=region,
delegated_management_subnet_id=get_subnet_for_region(region),
node_count=3,
sku="Standard_DS14_v2",
disk_sku="P50",
availability_zone=True
)
)
poller = client.cassandra_data_centers.begin_create_update(
"my-rg", "my-cluster", dc_name, datacenter_config
)
datacenter = poller.result()
print(f"Added data center in {region}: {datacenter.name}")
def get_subnet_for_region(region: str) -> str:
"""Get the appropriate subnet ID for a region."""
subnet_mapping = {
"West US": "/subscriptions/sub/resourceGroups/rg-west/providers/Microsoft.Network/virtualNetworks/vnet-west/subnets/cassandra",
"North Europe": "/subscriptions/sub/resourceGroups/rg-eu/providers/Microsoft.Network/virtualNetworks/vnet-eu/subnets/cassandra",
"Southeast Asia": "/subscriptions/sub/resourceGroups/rg-asia/providers/Microsoft.Network/virtualNetworks/vnet-asia/subnets/cassandra"
}
return subnet_mapping.get(region)Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-cosmosdb