azure-cosmos-db-py

仓库: sickn33/antigravity-awesome-skills
安装量: 50
排名: #14653

安装

npx skills add https://github.com/sickn33/antigravity-awesome-skills --skill azure-cosmos-db-py

Cosmos DB Service Implementation Build production-grade Azure Cosmos DB NoSQL services following clean code, security best practices, and TDD principles. Installation pip install azure-cosmos azure-identity Environment Variables COSMOS_ENDPOINT = https:// < account

.documents.azure.com:443/ COSMOS_DATABASE_NAME = < database-name

COSMOS_CONTAINER_ID

< container-id

For emulator only (not production)

COSMOS_KEY

< emulator-key

Authentication DefaultAzureCredential (preferred) : from azure . cosmos import CosmosClient from azure . identity import DefaultAzureCredential client = CosmosClient ( url = os . environ [ "COSMOS_ENDPOINT" ] , credential = DefaultAzureCredential ( ) ) Emulator (local development) : from azure . cosmos import CosmosClient client = CosmosClient ( url = "https://localhost:8081" , credential = os . environ [ "COSMOS_KEY" ] , connection_verify = False ) Architecture Overview ┌─────────────────────────────────────────────────────────────────┐ │ FastAPI Router │ │ - Auth dependencies (get_current_user, get_current_user_required) │ - HTTP error responses (HTTPException) │ └──────────────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────────────▼──────────────────────────────────┐ │ Service Layer │ │ - Business logic and validation │ │ - Document ↔ Model conversion │ │ - Graceful degradation when Cosmos unavailable │ └──────────────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────────────▼──────────────────────────────────┐ │ Cosmos DB Client Module │ │ - Singleton container initialization │ │ - Dual auth: DefaultAzureCredential (Azure) / Key (emulator) │ │ - Async wrapper via run_in_threadpool │ └─────────────────────────────────────────────────────────────────┘ Quick Start 1. Client Module Setup Create a singleton Cosmos client with dual authentication:

db/cosmos.py

from
azure
.
cosmos
import
CosmosClient
from
azure
.
identity
import
DefaultAzureCredential
from
starlette
.
concurrency
import
run_in_threadpool
_cosmos_container
=
None
def
_is_emulator_endpoint
(
endpoint
:
str
)
-
>
bool
:
return
"localhost"
in
endpoint
or
"127.0.0.1"
in
endpoint
async
def
get_container
(
)
:
global
_cosmos_container
if
_cosmos_container
is
None
:
if
_is_emulator_endpoint
(
settings
.
cosmos_endpoint
)
:
client
=
CosmosClient
(
url
=
settings
.
cosmos_endpoint
,
credential
=
settings
.
cosmos_key
,
connection_verify
=
False
)
else
:
client
=
CosmosClient
(
url
=
settings
.
cosmos_endpoint
,
credential
=
DefaultAzureCredential
(
)
)
db
=
client
.
get_database_client
(
settings
.
cosmos_database_name
)
_cosmos_container
=
db
.
get_container_client
(
settings
.
cosmos_container_id
)
return
_cosmos_container
Full implementation
See references/client-setup.md 2. Pydantic Model Hierarchy Use five-tier model pattern for clean separation: class ProjectBase ( BaseModel ) :

Shared fields

name : str = Field ( . . . , min_length = 1 , max_length = 200 ) class ProjectCreate ( ProjectBase ) :

Creation request

workspace_id : str = Field ( . . . , alias = "workspaceId" ) class ProjectUpdate ( BaseModel ) :

Partial updates (all optional)

name : Optional [ str ] = Field ( None , min_length = 1 ) class Project ( ProjectBase ) :

API response

id : str created_at : datetime = Field ( . . . , alias = "createdAt" ) class ProjectInDB ( Project ) :

Internal with docType

doc_type
:
str
=
"project"
3. Service Layer Pattern
class
ProjectService
:
def
_use_cosmos
(
self
)
-
>
bool
:
return
get_container
(
)
is
not
None
async
def
get_by_id
(
self
,
project_id
:
str
,
workspace_id
:
str
)
-
>
Project
|
None
:
if
not
self
.
_use_cosmos
(
)
:
return
None
doc
=
await
get_document
(
project_id
,
partition_key
=
workspace_id
)
if
doc
is
None
:
return
None
return
self
.
_doc_to_model
(
doc
)
Full patterns
See references/service-layer.md
Core Principles
Security Requirements
RBAC Authentication
Use
DefaultAzureCredential
in Azure — never store keys in code
Emulator-Only Keys
Hardcode the well-known emulator key only for local development
Parameterized Queries
Always use
@parameter
syntax — never string concatenation
Partition Key Validation
Validate partition key access matches user authorization
Clean Code Conventions
Single Responsibility
Client module handles connection; services handle business logic
Graceful Degradation
Services return
None
/
[]
when Cosmos unavailable
Consistent Naming
:
_doc_to_model()
,
_model_to_doc()
,
_use_cosmos()
Type Hints
Full typing on all public methods
CamelCase Aliases
Use Field(alias="camelCase") for JSON serialization TDD Requirements Write tests BEFORE implementation using these patterns: @pytest . fixture def mock_cosmos_container ( mocker ) : container = mocker . MagicMock ( ) mocker . patch ( "app.db.cosmos.get_container" , return_value = container ) return container @pytest . mark . asyncio async def test_get_project_by_id_returns_project ( mock_cosmos_container ) :

Arrange

mock_cosmos_container . read_item . return_value = { "id" : "123" , "name" : "Test" }

Act

result

await project_service . get_by_id ( "123" , "workspace-1" )

Assert

assert
result
.
id
==
"123"
assert
result
.
name
==
"Test"
Full testing guide
See references/testing.md Reference Files File When to Read references/client-setup.md Setting up Cosmos client with dual auth, SSL config, singleton pattern references/service-layer.md Implementing full service class with CRUD, conversions, graceful degradation references/testing.md Writing pytest tests, mocking Cosmos, integration test setup references/partitioning.md Choosing partition keys, cross-partition queries, move operations references/error-handling.md Handling CosmosResourceNotFoundError, logging, HTTP error mapping Template Files File Purpose assets/cosmos_client_template.py Ready-to-use client module assets/service_template.py Service class skeleton assets/conftest_template.py pytest fixtures for Cosmos mocking Quality Attributes (NFRs) Reliability Graceful degradation when Cosmos unavailable Retry logic with exponential backoff for transient failures Connection pooling via singleton pattern Security Zero secrets in code (RBAC via DefaultAzureCredential) Parameterized queries prevent injection Partition key isolation enforces data boundaries Maintainability Five-tier model pattern enables schema evolution Service layer decouples business logic from storage Consistent patterns across all entity services Testability Dependency injection via get_container() Easy mocking with module-level globals Clear separation enables unit testing without Cosmos Performance Partition key queries avoid cross-partition scans Async wrapping prevents blocking FastAPI event loop Minimal document conversion overhead When to Use This skill is applicable to execute the workflow or actions described in the overview.
← 返回排行榜