tessl/pypi-azure-cosmos

Microsoft Azure Cosmos Client Library for Python providing access to Azure Cosmos DB SQL API operations

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Client Operations

Name: tessl/pypi-azure-cosmos
Author: tessl

Account-level operations for Azure Cosmos DB including authentication, database management, and client configuration. The CosmosClient serves as the entry point for all Azure Cosmos DB operations.

Capabilities

Client Initialization

Initialize the Cosmos client with account endpoint and authentication credentials.

class CosmosClient:
    def __init__(self, url: str, credential: Union[str, Dict[str, Any], TokenCredential], consistency_level: str = None, **kwargs):
        """
        Initialize a new CosmosClient.
        
        Parameters:
        - url: Account endpoint URL
        - credential: Account key, resource token, token dictionary, or TokenCredential
        - consistency_level: Default consistency level for operations
        - connection_timeout: Connection timeout in seconds
        - request_timeout: Request timeout in seconds
        - connection_mode: Connection mode (Gateway or Direct)
        - proxy_config: Proxy configuration
        - ssl_config: SSL configuration
        - connection_verify: Enable SSL certificate verification
        - connection_cert: Client certificate file
        - retry_total: Total number of retries
        - retry_backoff_factor: Backoff factor for retries
        - retry_on_status_codes: HTTP status codes to retry on
        - enable_endpoint_discovery: Enable automatic endpoint discovery
        """

Alternative Client Creation

Create client from connection string for simplified configuration.

@classmethod
def from_connection_string(cls, conn_str: str, credential: str = None, consistency_level: str = None, **kwargs) -> CosmosClient:
    """
    Create a CosmosClient from connection string.
    
    Parameters:
    - conn_str: Azure Cosmos DB connection string
    - credential: Account key (if not in connection string)
    - consistency_level: Default consistency level
    - **kwargs: Additional client options
    
    Returns:
    CosmosClient instance
    """

Usage example:

# From connection string
client = CosmosClient.from_connection_string(
    conn_str="AccountEndpoint=https://account.documents.azure.com:443/;AccountKey=key==",
    consistency_level=ConsistencyLevel.Session
)

Database Management

Create, list, query, and delete databases within the account.

def create_database(self, id: str, populate_query_metrics: bool = None, offer_throughput: Union[int, ThroughputProperties] = None, **kwargs):
    """
    Create a new database.
    
    Parameters:
    - id: Database ID
    - populate_query_metrics: Include query metrics in response
    - offer_throughput: Provisioned throughput in RU/s
    - session_token: Session token for consistency
    - initial_headers: Custom headers
    - etag: ETag for conditional operations
    - match_condition: Match condition for conditional operations
    
    Returns:
    DatabaseProxy for the created database
    
    Raises:
    CosmosResourceExistsError: If database already exists
    """

def create_database_if_not_exists(self, id: str, populate_query_metrics: bool = None, offer_throughput: Union[int, ThroughputProperties] = None, **kwargs):
    """
    Create database if it doesn't exist, otherwise return existing.
    
    Parameters: Same as create_database
    
    Returns:
    DatabaseProxy for the database
    """

def get_database_client(self, database: str) -> DatabaseProxy:
    """
    Get a database client for operations on a specific database.
    
    Parameters:
    - database: Database ID or DatabaseProxy instance
    
    Returns:
    DatabaseProxy instance
    """

def list_databases(self, max_item_count: int = None, populate_query_metrics: bool = None, **kwargs):
    """
    List all databases in the account.
    
    Parameters:
    - max_item_count: Maximum number of items to return
    - populate_query_metrics: Include query metrics
    - session_token: Session token for consistency
    
    Returns:
    Iterable of database items
    """

def query_databases(self, query: str = None, parameters: list = None, enable_cross_partition_query: bool = None, max_item_count: int = None, populate_query_metrics: bool = None, **kwargs):
    """
    Query databases using SQL-like syntax.
    
    Parameters:
    - query: SQL query string
    - parameters: Query parameters as [{"name": "@param", "value": value}]
    - enable_cross_partition_query: Enable cross-partition queries
    - max_item_count: Maximum items per page
    - populate_query_metrics: Include query metrics
    - partition_key: Partition key value
    - session_token: Session token for consistency
    
    Returns:
    Iterable of query results
    """

def delete_database(self, database: str, populate_query_metrics: bool = None, **kwargs):
    """
    Delete a database.
    
    Parameters:
    - database: Database ID or DatabaseProxy instance  
    - populate_query_metrics: Include query metrics
    - session_token: Session token for consistency
    - etag: ETag for conditional operations
    - match_condition: Match condition for conditional operations
    
    Raises:
    CosmosResourceNotFoundError: If database doesn't exist
    """

Account Information

Retrieve account-level information and capabilities.

def get_database_account(self, **kwargs) -> DatabaseAccount:
    """
    Get database account information.
    
    Parameters:
    - session_token: Session token for consistency
    
    Returns:
    DatabaseAccount with account properties and capabilities
    """

Usage Examples

Basic Client Setup

from azure.cosmos import CosmosClient, ConsistencyLevel

# Initialize with explicit endpoint and key
client = CosmosClient(
    url="https://myaccount.documents.azure.com:443/",
    credential="myaccountkey==",
    consistency_level=ConsistencyLevel.Session
)

# Initialize with connection string
client = CosmosClient.from_connection_string(
    "AccountEndpoint=https://myaccount.documents.azure.com:443/;AccountKey=key=="
)

Database Operations

# Create a new database
try:
    database = client.create_database(
        id="ProductCatalog",
        offer_throughput=400
    )
    print(f"Created database: {database.id}")
except CosmosResourceExistsError:
    print("Database already exists")
    database = client.get_database_client("ProductCatalog")

# List all databases
databases = list(client.list_databases())
for db in databases:
    print(f"Database: {db['id']}")

# Query databases
results = list(client.query_databases(
    query="SELECT * FROM root r WHERE CONTAINS(r.id, @name)",
    parameters=[{"name": "@name", "value": "Product"}]
))

# Get account information
account_info = client.get_database_account()
print(f"Account consistency: {account_info.ConsistencyPolicy}")
print(f"Writable regions: {account_info.WritableLocations}")

Error Handling

from azure.cosmos.exceptions import (
    CosmosHttpResponseError,
    CosmosResourceExistsError,
    CosmosResourceNotFoundError
)

try:
    database = client.create_database("MyDatabase")
except CosmosResourceExistsError:
    print("Database already exists")
    database = client.get_database_client("MyDatabase")
except CosmosHttpResponseError as e:
    print(f"HTTP error occurred: {e.status_code} - {e.message}")

Install with Tessl CLI