Microsoft Azure Cosmos DB Management Client Library for Python
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Comprehensive management of SQL (Core) API resources including databases, containers, stored procedures, triggers, user-defined functions, client encryption keys, and RBAC role definitions. The SQL API is Cosmos DB's native API providing the richest feature set and maximum flexibility.
Create, configure, and manage SQL databases within Cosmos DB accounts.
def list_sql_databases(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[SqlDatabaseGetResults]:
"""
List SQL 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[SqlDatabaseGetResults]: Paginated list of SQL databases
"""
def get_sql_database(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> SqlDatabaseGetResults:
"""
Get properties of a SQL database.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
Returns:
SqlDatabaseGetResults: SQL database properties and configuration
"""
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]:
"""
Create or update a SQL 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 SQL database to create/update
- create_update_sql_database_parameters: Database configuration parameters
Returns:
LROPoller[SqlDatabaseGetResults]: Poller for monitoring operation progress
"""
def begin_delete_sql_database(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> LROPoller[None]:
"""
Delete a SQL 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 SQL database to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Manage SQL containers (collections) with advanced features like partitioning, indexing, and conflict resolution.
def list_sql_containers(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> ItemPaged[SqlContainerGetResults]:
"""
List SQL containers 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 SQL database
Returns:
ItemPaged[SqlContainerGetResults]: Paginated list of SQL containers
"""
def get_sql_container(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str
) -> SqlContainerGetResults:
"""
Get properties of a SQL container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
- container_name: Name of the SQL container
Returns:
SqlContainerGetResults: SQL container properties and configuration
"""
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]:
"""
Create or update a SQL container (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 SQL database
- container_name: Name of the SQL container to create/update
- create_update_sql_container_parameters: Container configuration parameters
Returns:
LROPoller[SqlContainerGetResults]: Poller for monitoring operation progress
"""
def begin_delete_sql_container(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str
) -> LROPoller[None]:
"""
Delete a SQL container (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 SQL database
- container_name: Name of the SQL container to delete
Returns:
LROPoller[None]: Poller for monitoring operation progress
"""Manage dedicated and shared throughput settings with support for autoscale and manual modes.
def get_sql_database_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> ThroughputSettingsGetResults:
"""
Get throughput settings for a SQL database.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
Returns:
ThroughputSettingsGetResults: Current throughput configuration
"""
def begin_update_sql_database_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str,
update_throughput_parameters: ThroughputSettingsUpdateParameters
) -> LROPoller[ThroughputSettingsGetResults]:
"""
Update throughput settings for a SQL 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 SQL database
- update_throughput_parameters: New throughput configuration
Returns:
LROPoller[ThroughputSettingsGetResults]: Poller for monitoring operation progress
"""
def get_sql_container_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str
) -> ThroughputSettingsGetResults:
"""
Get throughput settings for a SQL container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
- container_name: Name of the SQL container
Returns:
ThroughputSettingsGetResults: Current throughput configuration
"""
def begin_migrate_sql_database_to_autoscale(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> LROPoller[ThroughputSettingsGetResults]:
"""
Migrate SQL 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 SQL database
Returns:
LROPoller[ThroughputSettingsGetResults]: Poller for monitoring operation progress
"""
def begin_migrate_sql_database_to_manual_throughput(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> LROPoller[ThroughputSettingsGetResults]:
"""
Migrate SQL database 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 SQL database
Returns:
LROPoller[ThroughputSettingsGetResults]: Poller for monitoring operation progress
"""Manage server-side logic execution with stored procedures.
def list_sql_stored_procedures(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str
) -> ItemPaged[SqlStoredProcedureGetResults]:
"""
List SQL stored procedures in a container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
- container_name: Name of the SQL container
Returns:
ItemPaged[SqlStoredProcedureGetResults]: Paginated list of stored procedures
"""
def get_sql_stored_procedure(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str,
stored_procedure_name: str
) -> SqlStoredProcedureGetResults:
"""
Get properties of a SQL stored procedure.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
- container_name: Name of the SQL container
- stored_procedure_name: Name of the stored procedure
Returns:
SqlStoredProcedureGetResults: Stored procedure properties and body
"""
def begin_create_update_sql_stored_procedure(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str,
stored_procedure_name: str,
create_update_sql_stored_procedure_parameters: SqlStoredProcedureCreateUpdateParameters
) -> LROPoller[SqlStoredProcedureGetResults]:
"""
Create or update a SQL stored procedure (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 SQL database
- container_name: Name of the SQL container
- stored_procedure_name: Name of the stored procedure to create/update
- create_update_sql_stored_procedure_parameters: Stored procedure configuration
Returns:
LROPoller[SqlStoredProcedureGetResults]: Poller for monitoring operation progress
"""Manage triggers and user-defined functions for advanced data processing scenarios.
def list_sql_triggers(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str
) -> ItemPaged[SqlTriggerGetResults]:
"""
List SQL triggers in a container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
- container_name: Name of the SQL container
Returns:
ItemPaged[SqlTriggerGetResults]: Paginated list of triggers
"""
def list_sql_user_defined_functions(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str
) -> ItemPaged[SqlUserDefinedFunctionGetResults]:
"""
List SQL user-defined functions in a container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- database_name: Name of the SQL database
- container_name: Name of the SQL container
Returns:
ItemPaged[SqlUserDefinedFunctionGetResults]: Paginated list of UDFs
"""
def begin_create_update_sql_trigger(
self,
resource_group_name: str,
account_name: str,
database_name: str,
container_name: str,
trigger_name: str,
create_update_sql_trigger_parameters: SqlTriggerCreateUpdateParameters
) -> LROPoller[SqlTriggerGetResults]:
"""
Create or update a SQL trigger (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 SQL database
- container_name: Name of the SQL container
- trigger_name: Name of the trigger to create/update
- create_update_sql_trigger_parameters: Trigger configuration
Returns:
LROPoller[SqlTriggerGetResults]: Poller for monitoring operation progress
"""Manage role-based access control and client encryption keys for advanced security scenarios.
def list_sql_role_definitions(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[SqlRoleDefinitionGetResults]:
"""
List SQL role definitions for RBAC.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
Returns:
ItemPaged[SqlRoleDefinitionGetResults]: Paginated list of role definitions
"""
def get_sql_role_definition(
self,
resource_group_name: str,
account_name: str,
role_definition_id: str
) -> SqlRoleDefinitionGetResults:
"""
Get properties of a SQL role definition.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- role_definition_id: Unique identifier of the role definition
Returns:
SqlRoleDefinitionGetResults: Role definition properties and permissions
"""
def begin_create_update_sql_role_definition(
self,
resource_group_name: str,
account_name: str,
role_definition_id: str,
create_update_sql_role_definition_parameters: SqlRoleDefinitionCreateUpdateParameters
) -> LROPoller[SqlRoleDefinitionGetResults]:
"""
Create or update a SQL role definition (Long Running Operation).
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the Cosmos DB account
- role_definition_id: Unique identifier of the role definition
- create_update_sql_role_definition_parameters: Role definition configuration
Returns:
LROPoller[SqlRoleDefinitionGetResults]: Poller for monitoring operation progress
"""
def list_client_encryption_keys(
self,
resource_group_name: str,
account_name: str,
database_name: str
) -> ItemPaged[ClientEncryptionKeyGetResults]:
"""
List client encryption keys 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 SQL database
Returns:
ItemPaged[ClientEncryptionKeyGetResults]: Paginated list of encryption keys
"""from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.mgmt.cosmosdb.models import SqlDatabaseCreateUpdateParameters, SqlDatabaseResource, CreateUpdateOptions
from azure.identity import DefaultAzureCredential
client = CosmosDBManagementClient(DefaultAzureCredential(), "subscription-id")
# Define database configuration with shared throughput
database_params = SqlDatabaseCreateUpdateParameters(
resource=SqlDatabaseResource(id="products-db"),
options=CreateUpdateOptions(throughput=800) # Shared throughput
)
# Create database
poller = client.sql_resources.begin_create_update_sql_database(
"my-resource-group",
"my-cosmos-account",
"products-db",
database_params
)
database = poller.result()
print(f"Created database: {database.name}")from azure.mgmt.cosmosdb.models import (
SqlContainerCreateUpdateParameters,
SqlContainerResource,
ContainerPartitionKey,
IndexingPolicy,
IncludedPath,
ExcludedPath
)
# Define container configuration
container_resource = SqlContainerResource(
id="products",
partition_key=ContainerPartitionKey(
paths=["/category"],
kind="Hash"
),
indexing_policy=IndexingPolicy(
indexing_mode="Consistent",
automatic=True,
included_paths=[
IncludedPath(path="/*")
],
excluded_paths=[
ExcludedPath(path="/description/*")
]
)
)
container_params = SqlContainerCreateUpdateParameters(
resource=container_resource,
options=CreateUpdateOptions(throughput=400) # Dedicated throughput
)
# Create container
poller = client.sql_resources.begin_create_update_sql_container(
"my-resource-group",
"my-cosmos-account",
"products-db",
"products",
container_params
)
container = poller.result()
print(f"Created container: {container.name} with partition key: {container.resource.partition_key.paths}")from azure.mgmt.cosmosdb.models import ThroughputSettingsUpdateParameters, ThroughputSettingsResource, AutoscaleSettings
# Update to autoscale throughput
autoscale_params = ThroughputSettingsUpdateParameters(
resource=ThroughputSettingsResource(
autoscale_settings=AutoscaleSettings(max_throughput=4000)
)
)
poller = client.sql_resources.begin_update_sql_database_throughput(
"my-resource-group",
"my-cosmos-account",
"products-db",
autoscale_params
)
throughput = poller.result()
print(f"Updated to autoscale with max RU/s: {throughput.resource.autoscale_settings.max_throughput}")
# Migrate to manual throughput
poller = client.sql_resources.begin_migrate_sql_database_to_manual_throughput(
"my-resource-group",
"my-cosmos-account",
"products-db"
)
throughput = poller.result()
print(f"Migrated to manual throughput: {throughput.resource.throughput} RU/s")class SqlDatabaseGetResults:
"""SQL database properties and configuration."""
id: str
name: str
type: str
resource: SqlDatabaseGetPropertiesResource
options: SqlDatabaseGetPropertiesOptions
class SqlDatabaseResource:
"""SQL database resource definition."""
id: str # Database name
class SqlContainerGetResults:
"""SQL container properties and configuration."""
id: str
name: str
type: str
resource: SqlContainerGetPropertiesResource
options: SqlContainerGetPropertiesOptions
class SqlContainerResource:
"""SQL container resource definition."""
id: str # Container name
partition_key: ContainerPartitionKey
indexing_policy: IndexingPolicy
unique_key_policy: UniqueKeyPolicy
conflict_resolution_policy: ConflictResolutionPolicy
default_ttl: int
analytical_storage_ttl: int
class ContainerPartitionKey:
"""Partition key configuration for containers."""
paths: List[str] # Partition key paths (e.g., ["/category"])
kind: str # "Hash" or "Range"
version: int # Partition key version (1 or 2)
class IndexingPolicy:
"""Indexing policy configuration for containers."""
indexing_mode: str # "Consistent", "Lazy", "None"
automatic: bool
included_paths: List[IncludedPath]
excluded_paths: List[ExcludedPath]
composite_indexes: List[List[CompositePath]]
spatial_indexes: List[SpatialSpec]
class ThroughputSettingsGetResults:
"""Throughput settings and configuration."""
id: str
name: str
type: str
resource: ThroughputSettingsGetPropertiesResource
class ThroughputSettingsResource:
"""Throughput resource definition."""
throughput: int # Manual throughput in RU/s
autoscale_settings: AutoscaleSettings # Autoscale configuration
minimum_throughput: int # Minimum allowed throughput
class AutoscaleSettings:
"""Autoscale throughput configuration."""
max_throughput: int # Maximum RU/s for autoscale
class SqlStoredProcedureGetResults:
"""SQL stored procedure properties and body."""
id: str
name: str
type: str
resource: SqlStoredProcedureGetPropertiesResource
class SqlStoredProcedureResource:
"""SQL stored procedure resource definition."""
id: str # Stored procedure name
body: str # JavaScript code body
class SqlRoleDefinitionGetResults:
"""SQL role definition for RBAC."""
id: str
name: str
type: str
role_name: str
type_value: str # "BuiltInRole" or "CustomRole"
assignable_scopes: List[str]
permissions: List[Permission]
class Permission:
"""Permission definition for RBAC roles."""
data_actions: List[str] # Allowed data plane actions
not_data_actions: List[str] # Denied data plane actionsInstall with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-cosmosdb