DatabaseProxy Class

Create a new container with the given ID (name).

If a container with the given ID already exists, a CosmosResourceExistsError is raised.

async create_container(id: str, partition_key: PartitionKey, *, indexing_policy: dict[str, str] | None = None, default_ttl: int | None = None, offer_throughput: int | ThroughputProperties | None = None, unique_key_policy: dict[str, str] | None = None, conflict_resolution_policy: dict[str, str] | None = None, initial_headers: dict[str, str] | None = None, computed_properties: list[dict[str, str]] | None = None, analytical_storage_ttl: int | None = None, vector_embedding_policy: dict[str, Any] | None = None, change_feed_policy: dict[str, Any] | None = None, full_text_policy: dict[str, Any] | None = None, return_properties: Literal[False] = False, **kwargs: Any) -> ContainerProxy

Parameters

Keyword-Only Parameters

Returns

Exceptions

Examples

Create a container with default settings:

           container_name = "products"
           try:
               container = await database.create_container(
                   id=container_name, partition_key=PartitionKey(path="/productName")
               )
           except exceptions.CosmosResourceExistsError:
               container = database.get_container_client(container_name)

Create a container with specific settings; in this case, a custom partition key:

           customer_container_name = "customers"
           try:
               customer_container = await database.create_container(
                   id=customer_container_name,
                   partition_key=PartitionKey(path="/city"),
                   default_ttl=200,
               )
           except exceptions.CosmosResourceExistsError:
               customer_container = database.get_container_client(customer_container_name)

Create a container if it does not exist already.

If the container already exists, the existing settings are returned. Note: it does not check or update the existing container settings or offer throughput if they differ from what was passed into the method.

async create_container_if_not_exists(id: str, partition_key: PartitionKey, *, indexing_policy: dict[str, str] | None = None, default_ttl: int | None = None, offer_throughput: int | ThroughputProperties | None = None, unique_key_policy: dict[str, str] | None = None, conflict_resolution_policy: dict[str, str] | None = None, initial_headers: dict[str, str] | None = None, computed_properties: list[dict[str, str]] | None = None, analytical_storage_ttl: int | None = None, vector_embedding_policy: dict[str, Any] | None = None, change_feed_policy: dict[str, Any] | None = None, full_text_policy: dict[str, Any] | None = None, return_properties: Literal[False] = False, **kwargs: Any) -> ContainerProxy

Parameters

Keyword-Only Parameters

Returns

Exceptions

Create a new user in the container.

To update or replace an existing user, use the <xref:ContainerProxy.upsert_user> method.

async create_user(body: dict[str, Any], **kwargs: Any) -> UserProxy

Parameters

Keyword-Only Parameters

Returns

Exceptions

Examples

Create a database user:

           try:
               await database.create_user(dict(id="Walter Harp"))
               print("Created user Walter Harp.")
           except exceptions.CosmosResourceExistsError:
               print("A user with that ID already exists.")
           except exceptions.CosmosHttpResponseError as failure:
               print("Failed to create user. Status code:{}".format(failure.status_code))

Delete a container.

async delete_container(container: str | ContainerProxy | Mapping[str, Any], *, initial_headers: dict[str, str] | None = None, **kwargs: Any) -> None

Parameters

Keyword-Only Parameters

Returns

Exceptions

Delete the specified user from the container.

async delete_user(user: str | UserProxy | Mapping[str, Any], **kwargs: Any) -> None

Parameters

Keyword-Only Parameters

Returns

Exceptions

Get a ContainerProxy for a container with specified ID (name).

get_container_client(container: str | ContainerProxy | dict[str, Any]) -> ContainerProxy

Parameters

Returns

Examples

Get an existing container, handling a failure if encountered:

           database = client.get_database_client(database_name)
           container_client = database.get_container_client(container_name)

Get the ThroughputProperties object for this database.

If no ThroughputProperties already exists for the database, an exception is raised.

async get_throughput(*, response_hook: Callable[[Mapping[str, Any], list[dict[str, Any]]], None] | None = None, **kwargs: Any) -> ThroughputProperties

Keyword-Only Parameters

Returns

Exceptions

Get a UserProxy for a user with specified ID.

get_user_client(user: str | UserProxy | Mapping[str, Any]) -> UserProxy

Parameters

Returns

List the containers in the database.

list_containers(*, max_item_count: int | None = None, initial_headers: dict[str, str] | None = None, response_hook: Callable[[Mapping[str, Any], AsyncItemPaged[dict[str, Any]]], None] | None = None, **kwargs) -> AsyncItemPaged[dict[str, Any]]

Keyword-Only Parameters

Returns

Examples

List all containers in the database:

           database = client.get_database_client(database_name)
           async for container_c in database.list_containers():
               print("Container ID: {}".format(container_c['id']))

List all the users in the container.

list_users(*, max_item_count: int | None = None, response_hook: Callable[[Mapping[str, Any], AsyncItemPaged[dict[str, Any]]], None] | None = None, **kwargs: Any) -> AsyncItemPaged[dict[str, Any]]

Keyword-Only Parameters

Returns

List the properties for containers in the current database.

query_containers(query: str, *, parameters: list[dict[str, Any]] | None = None, max_item_count: int | None = None, initial_headers: dict[str, str] | None = None, response_hook: Callable[[Mapping[str, Any], AsyncItemPaged[dict[str, Any]]], None] | None = None, **kwargs: Any) -> AsyncItemPaged[dict[str, Any]]

Parameters

Keyword-Only Parameters

Returns

Return all users matching the given query.

query_users(query: str, *, parameters: list[dict[str, Any]] | None = None, max_item_count: int | None = None, response_hook: Callable[[Mapping[str, Any], AsyncItemPaged[dict[str, Any]]], None] | None = None, **kwargs: Any) -> AsyncItemPaged[dict[str, Any]]

Parameters

Keyword-Only Parameters

Returns

Read the database properties.

async read(*, initial_headers: dict[str, str] | None = None, **kwargs: Any) -> CosmosDict

Keyword-Only Parameters

Returns

Exceptions

Reset the properties of the container.

Property changes are persisted immediately. Any properties not specified will be reset to their default values.

async replace_container(container: str | ContainerProxy | Mapping[str, Any], partition_key: PartitionKey, *, indexing_policy: dict[str, str] | None = None, default_ttl: int | None = None, conflict_resolution_policy: dict[str, str] | None = None, initial_headers: dict[str, str] | None = None, analytical_storage_ttl: int | None = None, computed_properties: list[dict[str, str]] | None = None, full_text_policy: dict[str, Any] | None = None, return_properties: Literal[False] = False, vector_embedding_policy: dict[str, Any] | None = None, **kwargs: Any) -> ContainerProxy

Parameters

Keyword-Only Parameters

Returns

Exceptions

Examples

Reset the TTL property on a container, and display the updated properties:

           # Set the TTL on the container to 3600 seconds (one hour)
           await database.replace_container(container, partition_key=PartitionKey(path='/productName'), default_ttl=3600)

           # Display the new TTL setting for the container
           container_props = await database.get_container_client(container_name).read()
           print("New container TTL: {}".format(json.dumps(container_props['defaultTtl'])))

Replace the database-level throughput.

If no ThroughputProperties already exist for the database, an exception is raised.

async replace_throughput(throughput: int | ThroughputProperties, **kwargs: Any) -> ThroughputProperties

Parameters

Keyword-Only Parameters

Returns

Exceptions

Replaces the specified user if it exists in the container.

async replace_user(user: str | UserProxy | Mapping[str, Any], body: dict[str, Any], **kwargs: Any) -> UserProxy

Parameters

Keyword-Only Parameters

Returns

Exceptions

Insert or update the specified user.

If the user already exists in the container, it is replaced. If the user does not already exist, it is inserted.

async upsert_user(body: dict[str, Any], **kwargs: Any) -> UserProxy

Parameters

Keyword-Only Parameters

Returns

Exceptions