Coverage for src/lexigram/admin/data/read_only.py: 59%

1"""Read-only data source mixin for lexigram-admin.""" 

2 

3from __future__ import annotations 

4 

5from abc import ABC, abstractmethod 

6from typing import Any, Generic, TypeVar 

7 

8from lexigram.admin.data.data_source import QueryResult 

9from lexigram.admin.data.query import QuerySpec 

10 

11T = TypeVar("T") 

12 

13 

14class ReadOnlyError(Exception): 

15 """Raised when a write is attempted on a read-only data source.""" 

16 

17 

18class ReadOnlyDataSource(ABC, Generic[T]): 

19 """ABC for data sources that support reads only. 

20 

21 Subclasses implement find_one, find_many, count. 

22 All write methods raise ReadOnlyError. 

23 """ 

24 

25 @abstractmethod 

26 async def find_one(self, item_id: Any) -> T | None: ... 

27 

28 @abstractmethod 

29 async def find_many(self, query: QuerySpec) -> QueryResult[T]: ... 

30 

31 @abstractmethod 

32 async def count(self, query: QuerySpec) -> int: ... 

33 

34 def _read_only(self) -> ReadOnlyError: 

35 return ReadOnlyError(f"{type(self).__name__} is read-only") 

36 

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

38 raise self._read_only() 

39 

40 async def update(self, item_id: Any, data: dict[str, Any]) -> T: 

41 raise self._read_only() 

42 

43 async def delete(self, item_id: Any) -> bool: 

44 raise self._read_only() 

45 

46 async def bulk_create(self, items: list[dict[str, Any]]) -> list[T]: 

47 raise self._read_only() 

48 

49 async def bulk_update(self, ids: list[Any], data: dict[str, Any]) -> int: 

50 raise self._read_only() 

51 

52 async def bulk_delete(self, ids: list[Any]) -> int: 

53 raise self._read_only()