Coverage for src/lexigram/admin/data/adapters/repository/data_source.py: 96%

1"""AdminRepositoryProtocol-backed IDataSource adapter.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, Generic 

6 

7from lexigram.admin.data.adapters.repository.types import T 

8from lexigram.admin.data.data_source import IDataSource, QueryResult 

9from lexigram.admin.data.query import QuerySpec 

10from lexigram.contracts.admin.repository import AdminRepositoryProtocol 

11from lexigram.di.decorators import inject 

12from lexigram.logging import get_logger 

13 

14logger = get_logger(__name__) 

15 

16 

17@inject 

18class RepositoryDataSource(IDataSource[T], Generic[T]): 

19 """IDataSource adapter for AdminRepositoryProtocol repositories. 

20 

21 Translates QuerySpec into AdminRepositoryProtocol calls using 

22 QuerySpec.resolved_sort and QuerySpec.to_repository_filters(). 

23 """ 

24 

25 def __init__(self, repository: AdminRepositoryProtocol[T]) -> None: 

26 self._repo = repository 

27 

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

29 return await self._repo.find_by_id(item_id) 

30 

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

32 sort_field, sort_order = query.resolved_sort 

33 filters = query.to_repository_filters() 

34 

35 items = await self._repo.find_many( 

36 offset=query.offset, 

37 limit=query.per_page, 

38 order_by=[(sort_field, sort_order)] if sort_field else None, 

39 filters=filters, 

40 search=query.search or None, 

41 search_fields=query.search_fields or None, 

42 load=query.include or None, 

43 ) 

44 

45 total = await self.count(query) 

46 

47 return QueryResult( 

48 items=items, 

49 total=total, 

50 page=query.page, 

51 per_page=query.per_page, 

52 has_next=(query.offset + query.per_page) < total, 

53 has_prev=query.page > 1, 

54 ) 

55 

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

57 return await self._repo.count( 

58 filters=query.to_repository_filters(), 

59 search=query.search or None, 

60 search_fields=query.search_fields or None, 

61 ) 

62 

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

64 return await self._repo.create(data) 

65 

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

67 return await self._repo.update(item_id, data) 

68 

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

70 return await self._repo.delete(item_id) 

71 

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

73 return [await self._repo.create(item) for item in items] 

74 

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

76 count = 0 

77 for item_id in ids: 

78 try: 

79 await self._repo.update(item_id, data) 

80 count += 1 

81 except Exception: 

82 logger.warning("bulk_update: skipped id=%s", item_id) 

83 return count 

84 

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

86 count = 0 

87 for item_id in ids: 

88 if await self._repo.delete(item_id): 

89 count += 1 

90 return count