Skip to content

documentdb

azure_bootstrap.documentdb

MongoDB / Cosmos DB document database access.

Functions:

Name Description
mongo_client_from_env

Build a pymongo MongoClient from env.
documentdb_health

Return {status, latency_ms} for health probes.

mongo_client_from_env 

mongo_client_from_env(*, uri_env: str = 'NOSQL_URI') -> Any

Build a pymongo MongoClient from env.

Source code in azure_bootstrap/documentdb/__init__.py 
14
15
16
17
18
19
def mongo_client_from_env(*, uri_env: str = "NOSQL_URI") -> Any:
    """Build a pymongo ``MongoClient`` from env."""
    import pymongo  # type: ignore[import-untyped]

    uri = require_env(uri_env)
    return pymongo.MongoClient(uri, serverSelectionTimeoutMS=5000)

documentdb_health 

documentdb_health(*, uri_env: str = 'NOSQL_URI', database_env: str = 'NOSQL_DATABASE') -> dict[str, Any]

Return {status, latency_ms} for health probes.

Source code in azure_bootstrap/documentdb/__init__.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def documentdb_health(
    *, uri_env: str = "NOSQL_URI", database_env: str = "NOSQL_DATABASE"
) -> dict[str, Any]:
    """Return ``{status, latency_ms}`` for health probes."""
    start = time.perf_counter()
    try:
        client = mongo_client_from_env(uri_env=uri_env)
        db_name = fail_open_env(database_env) or "admin"
        client[db_name].command("ping")
        latency = (time.perf_counter() - start) * 1000
        client.close()
        return {"status": "ok", "latency_ms": round(latency, 2)}
    except Exception as exc:
        return {"status": "error", "latency_ms": None, "error": str(exc)}