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

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

4 

5if TYPE_CHECKING: 

6 from lexigram.contracts.data import QueryResult 

7 

8 

9@runtime_checkable 

10class DataSourceProtocol(Protocol): 

11 """Protocol for admin data sources with full CRUD support.""" 

12 

13 async def get_data( 

14 self, 

15 filters: dict[str, Any] | None = None, 

16 sort: list[dict[str, str]] | None = None, 

17 limit: int | None = None, 

18 offset: int | None = None, 

19 ) -> QueryResult: ... 

20 

21 async def get_record_count(self, filters: dict[str, Any] | None = None) -> int: ... 

22 

23 async def create(self, data: dict[str, Any]) -> Any: ... 

24 

25 async def find_one(self, record_id: Any) -> Any | None: ... 

26 

27 async def find_many( 

28 self, 

29 filters: dict[str, Any] | None = None, 

30 sort: list[dict[str, str]] | None = None, 

31 limit: int | None = None, 

32 offset: int | None = None, 

33 ) -> list[Any]: ... 

34 

35 async def update(self, record_id: Any, data: dict[str, Any]) -> Any: ... 

36 

37 async def delete(self, record_id: Any) -> bool: ... 

38 

39 async def bulk_delete(self, filters: dict[str, Any]) -> int: ... 

40 

41 

42__all__ = ["DataSourceProtocol"]