Coverage for src / lexigram / contracts / data / nosql / nosql_repository.py: 0%

1"""NoSQL repository protocols for document-oriented data access. 

2 

3Provides a generic ``DocumentRepositoryProtocol[T]`` that mirrors 

4:class:`~lexigram.contracts.data.repository.RepositoryProtocol` 

5but uses document semantics (filter dicts, no SQL assumptions). 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable 

11 

12if TYPE_CHECKING: 

13 import builtins 

14 

15 from lexigram.contracts.domain.specification import SpecificationProtocol 

16 

17T = TypeVar("T") 

18 

19 

20@runtime_checkable 

21class DocumentRepositoryProtocol(Protocol[T]): 

22 """Repository protocol for document-oriented storage. 

23 

24 Mirrors ``RepositoryProtocol`` but uses document semantics: 

25 

26 - No SQL ``table_name`` / ``key_field`` assumptions 

27 - Filter expressions instead of WHERE clauses 

28 - Aggregation pipeline support 

29 """ 

30 

31 async def get(self, document_id: str) -> T | None: 

32 """Retrieve a document by its ID.""" 

33 ... 

34 

35 async def list( 

36 self, 

37 skip: int = 0, 

38 limit: int = 100, 

39 *, 

40 sort: list[tuple[str, int]] | None = None, 

41 **filters: Any, 

42 ) -> list[T]: 

43 """List documents with pagination and optional filters.""" 

44 ... 

45 

46 async def find_by_filter(self, filter: dict[str, Any]) -> builtins.list[T]: 

47 """Find documents matching a raw filter expression.""" 

48 ... 

49 

50 async def find_by_spec(self, spec: SpecificationProtocol[T]) -> builtins.list[T]: 

51 """Find documents matching a specification.""" 

52 ... 

53 

54 async def count(self, **filters: Any) -> int: 

55 """Count documents matching filters.""" 

56 ... 

57 

58 async def save(self, entity: T) -> T: 

59 """Insert or update a document (upsert semantics).""" 

60 ... 

61 

62 async def delete(self, document_id: str) -> bool: 

63 """Delete a document by ID.""" 

64 ... 

65 

66 async def save_many(self, entities: builtins.list[T]) -> builtins.list[T]: 

67 """Bulk insert / update documents.""" 

68 ... 

69 

70 async def delete_many(self, document_ids: builtins.list[str]) -> int: 

71 """Bulk delete documents by IDs.""" 

72 ... 

73 

74 

75__all__ = ["DocumentRepositoryProtocol", "T"]