Microsoft Azure Cosmos DB Management Client Library for Python
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Core database account lifecycle operations for Azure Cosmos DB. These operations manage the top-level Cosmos DB account resource that hosts all databases and containers, including creation, configuration, scaling, failover management, key rotation, and monitoring.
Create, retrieve, update, and delete Cosmos DB database accounts with comprehensive configuration options.
def get(self, resource_group_name: str, account_name: str) -> DatabaseAccountGetResults:
"""
Get the properties of an existing database account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
DatabaseAccountGetResults: Complete account information
"""
def begin_create_or_update(
self,
resource_group_name: str,
account_name: str,
create_update_parameters: DatabaseAccountCreateUpdateParameters
) -> LROPoller[DatabaseAccountGetResults]:
"""
Create or update a database account (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account to create/update
- create_update_parameters: Account configuration parameters
Returns:
LROPoller[DatabaseAccountGetResults]: Poller for monitoring operation progress
"""
def begin_update(
self,
resource_group_name: str,
account_name: str,
update_parameters: DatabaseAccountUpdateParameters
) -> LROPoller[DatabaseAccountGetResults]:
"""
Update properties of an existing database account (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account to update
- update_parameters: Account update parameters
Returns:
LROPoller[DatabaseAccountGetResults]: Poller for monitoring operation progress
"""
def begin_delete(
self,
resource_group_name: str,
account_name: str
) -> LROPoller[None]:
"""
Delete a database account (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""List and discover Cosmos DB accounts across subscriptions and resource groups.
def list(self) -> ItemPaged[DatabaseAccountGetResults]:
"""
List all database accounts in the subscription.
Returns:
ItemPaged[DatabaseAccountGetResults]: Paginated list of all accounts
"""
def list_by_resource_group(
self,
resource_group_name: str
) -> ItemPaged[DatabaseAccountGetResults]:
"""
List all database accounts in a resource group.
Parameters:
- resource_group_name: Name of the resource group
Returns:
ItemPaged[DatabaseAccountGetResults]: Paginated list of accounts in resource group
"""
def check_name_exists(self, account_name: str) -> bool:
"""
Check if a database account name is available.
Parameters:
- account_name: Name to check for availability
Returns:
bool: True if name exists, False if available
"""Manage access keys for authentication and security operations.
def list_keys(
self,
resource_group_name: str,
account_name: str
) -> DatabaseAccountListKeysResult:
"""
List access keys for a database account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
DatabaseAccountListKeysResult: Primary and secondary keys
"""
def list_read_only_keys(
self,
resource_group_name: str,
account_name: str
) -> DatabaseAccountListReadOnlyKeysResult:
"""
List read-only access keys for a database account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
DatabaseAccountListReadOnlyKeysResult: Read-only primary and secondary keys
"""
def list_connection_strings(
self,
resource_group_name: str,
account_name: str
) -> DatabaseAccountListConnectionStringsResult:
"""
List connection strings for a database account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
DatabaseAccountListConnectionStringsResult: Connection strings for different APIs
"""
def begin_regenerate_key(
self,
resource_group_name: str,
account_name: str,
key_to_regenerate: DatabaseAccountRegenerateKeyParameters
) -> LROPoller[None]:
"""
Regenerate an access key for a database account (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- key_to_regenerate: Specifies which key to regenerate
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Manage multi-region deployments, failover priorities, and region operations.
def begin_failover_priority_change(
self,
resource_group_name: str,
account_name: str,
failover_parameters: FailoverPolicies
) -> LROPoller[None]:
"""
Change failover priorities for multi-region accounts (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- failover_parameters: New failover priority configuration
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""
def begin_offline_region(
self,
resource_group_name: str,
account_name: str,
region_parameter_for_offline: RegionForOnlineOffline
) -> LROPoller[None]:
"""
Take a region offline for maintenance (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- region_parameter_for_offline: Region to take offline
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""
def begin_online_region(
self,
resource_group_name: str,
account_name: str,
region_parameter_for_online: RegionForOnlineOffline
) -> LROPoller[None]:
"""
Bring a region back online (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- region_parameter_for_online: Region to bring online
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Get metrics, usage statistics, and performance data for database accounts.
def list_metrics(
self,
resource_group_name: str,
account_name: str,
filter: str
) -> ItemPaged[Metric]:
"""
Get metrics for a database account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- filter: OData filter for metrics (e.g., timespan, metric names)
Returns:
ItemPaged[Metric]: Paginated list of metrics
"""
def list_usages(
self,
resource_group_name: str,
account_name: str,
filter: Optional[str] = None
) -> ItemPaged[Usage]:
"""
Get usage statistics for a database account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- filter: Optional OData filter for usage data
Returns:
ItemPaged[Usage]: Paginated list of usage statistics
"""
def list_metric_definitions(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[MetricDefinition]:
"""
Get available metric definitions for a database account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
ItemPaged[MetricDefinition]: Available metric definitions
"""from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.mgmt.cosmosdb.models import DatabaseAccountCreateUpdateParameters, Location
from azure.identity import DefaultAzureCredential
client = CosmosDBManagementClient(DefaultAzureCredential(), "subscription-id")
# Define account configuration
create_params = DatabaseAccountCreateUpdateParameters(
location="East US",
locations=[
Location(location_name="East US", failover_priority=0, is_zone_redundant=False),
Location(location_name="West US", failover_priority=1, is_zone_redundant=False)
],
kind="GlobalDocumentDB",
consistency_policy={
"default_consistency_level": "Session"
},
enable_automatic_failover=True,
enable_multiple_write_locations=False
)
# Create account (Long Running Operation)
poller = client.database_accounts.begin_create_or_update(
"my-resource-group",
"my-cosmos-account",
create_params
)
# Wait for completion
account = poller.result()
print(f"Created account: {account.name} in regions: {[loc.location_name for loc in account.locations]}")# List current keys
keys = client.database_accounts.list_keys("resource-group", "account-name")
print(f"Primary Key: {keys.primary_master_key}")
print(f"Secondary Key: {keys.secondary_master_key}")
# Regenerate secondary key
from azure.mgmt.cosmosdb.models import DatabaseAccountRegenerateKeyParameters
regenerate_params = DatabaseAccountRegenerateKeyParameters(key_kind="secondary")
poller = client.database_accounts.begin_regenerate_key(
"resource-group", "account-name", regenerate_params
)
poller.result() # Wait for completion
# Get connection strings
connection_strings = client.database_accounts.list_connection_strings(
"resource-group", "account-name"
)
for conn_str in connection_strings.connection_strings:
print(f"{conn_str.description}: {conn_str.connection_string}")class DatabaseAccountGetResults:
"""Complete database account information including properties, settings, and status."""
id: str
name: str
type: str
location: str
tags: Dict[str, str]
kind: str # "GlobalDocumentDB", "MongoDB", "Parse"
provisioning_state: str
document_endpoint: str
locations: List[Location]
read_locations: List[Location]
write_locations: List[Location]
consistency_policy: ConsistencyPolicy
capabilities: List[Capability]
enable_automatic_failover: bool
enable_multiple_write_locations: bool
backup_policy: BackupPolicy
class DatabaseAccountCreateUpdateParameters:
"""Parameters for creating or updating a database account."""
location: str
tags: Dict[str, str]
kind: str
locations: List[Location]
consistency_policy: ConsistencyPolicy
capabilities: List[Capability]
enable_automatic_failover: bool
enable_multiple_write_locations: bool
backup_policy: BackupPolicy
public_network_access: str
network_acl_bypass: str
ip_rules: List[IpAddressOrRange]
virtual_network_rules: List[VirtualNetworkRule]
class DatabaseAccountListKeysResult:
"""Database account access keys."""
primary_master_key: str
secondary_master_key: str
primary_readonly_master_key: str
secondary_readonly_master_key: str
class Location:
"""Azure region configuration for database account."""
location_name: str
failover_priority: int
is_zone_redundant: bool
class ConsistencyPolicy:
"""Database consistency policy settings."""
default_consistency_level: str # "Eventual", "Session", "BoundedStaleness", "Strong", "ConsistentPrefix"
max_staleness_prefix: int
max_interval_in_seconds: intInstall with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-cosmosdb