appfx-cosmosdb 0.1.0

Python helpers for Azure Cosmos DB SQL and MongoDB APIs.

appfx-cosmosdb

appfx-cosmosdb provides type-safe, async repository helpers for Azure Cosmos DB using either the SQL/Core API or the MongoDB API. The package keeps the two API surfaces explicit so applications import only the implementation they use.

Package identity

Purpose	Name
Distribution	appfx-cosmosdb
SQL/Core API namespace	appfx.cosmosdb.sql
MongoDB API namespace	appfx.cosmosdb.mongo

The top-level appfx.cosmosdb package is not the public home for repository or entity classes. Import from appfx.cosmosdb.sql or appfx.cosmosdb.mongo directly.

Installation

Install only the dependencies for the Cosmos DB API you use:

python -m pip install "appfx-cosmosdb[sql]"
python -m pip install "appfx-cosmosdb[mongo]"

Install both SQL and MongoDB dependencies when one application uses both APIs:

python -m pip install "appfx-cosmosdb[all]"

For local development:

python -m pip install --upgrade pip
python -m pip install -e ".[dev,all]"

SQL/Core API quick start

from appfx.cosmosdb.sql import RepositoryBase, RootEntityBase, SortDirection, SortField


class Customer(RootEntityBase["Customer", str]):
    name: str
    email: str
    is_active: bool = True


class CustomerRepository(RepositoryBase[Customer, str]):
    def __init__(self, connection_string: str, database_name: str):
        super().__init__(
            connection_string=connection_string,
            database_name=database_name,
            container_name="customers",
        )


async with CustomerRepository(connection_string, "app") as repo:
    await repo.add_async(Customer(id="customer-1", name="Ada", email="[e-mail]"))

    active_customers = await repo.find_async(
        {"is_active": True},
        sort_fields=[SortField("name", SortDirection.ASCENDING)],
    )

SQL entities automatically include a computed /_partitionKey value derived from the entity id. Use raw SQL methods for projections, aggregations, and other queries that are clearer in Cosmos DB SQL syntax.

MongoDB API quick start

from appfx.cosmosdb.mongo import RepositoryBase, RootEntityBase, SortDirection, SortField
from pydantic import Field


class Customer(RootEntityBase["Customer", str]):
    name: str
    email: str
    tags: list[str] = Field(default_factory=list)


class CustomerRepository(RepositoryBase[Customer, str]):
    def __init__(self, connection_string: str, database_name: str):
        super().__init__(
            connection_string=connection_string,
            database_name=database_name,
            collection_name="customers",
        )


async with CustomerRepository(connection_string, "app") as repo:
    await repo.add_async(Customer(id="customer-1", name="Ada", email="[e-mail]"))

    premium_customers = await repo.find_async(
        {"tags": {"$in": ["premium"]}},
        sort_fields=[SortField("name", SortDirection.ASCENDING)],
    )

Mongo repositories pass predicate dictionaries through to MongoDB-style query operators, including array and regular-expression operators supported by the driver and Cosmos DB MongoDB API.

Core concepts

• Entity models inherit from RootEntityBase["EntityName", KeyType] for independently stored documents and from EntityBase for nested models.
• Repositories inherit from the API-specific RepositoryBase[TEntity, TKey] and centralize CRUD, query, count, pagination, and bulk-delete operations.
• Sort helpers are public from both API namespaces: SortField and SortDirection.
• Resource cleanup should use async with repo: or await repo.close().

Documentation

• Documentation index
• API overview
• SQL/Core API reference
• MongoDB API reference
• Hands-on guide
• Standalone examples

Testing

The default test suite is unit-test focused:

python -m pytest

Live Cosmos DB tests require explicit integration setup, including a Cosmos DB account, API-specific connection settings, and any required Azure RBAC or network access. Do not put credentials in source-controlled files.

Development validation

python -m pytest
python -m pytest --cov
python -m ruff check .
python -m ruff format --check .
python -m mypy
python -m build
python -m twine check dist/*

License and provenance

This package is licensed under the MIT License. Portions of the documentation were adapted from earlier Cosmos DB helper documentation that was also MIT licensed; obsolete package names, repository URLs, and legacy branding were removed during the appfx-cosmosdb migration.