Coverage for src/lexigram/admin/data/adapters/export_adapter.py: 88%

1"""Adapter to make IDataSource compatible with ExportService.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.admin.data.query import QuerySpec 

8from lexigram.admin.services.export import IExportDataSource 

9from lexigram.di.decorators import inject 

10 

11if TYPE_CHECKING: 

12 from lexigram.admin.data.data_source import IDataSource 

13 

14 

15@inject 

16class ExportDataSourceAdapter(IExportDataSource): 

17 """Bridges IDataSource to IExportDataSource for ExportService.""" 

18 

19 def __init__(self, data_source: IDataSource) -> None: 

20 """Initialize with a new-style data source.""" 

21 self.data_source = data_source 

22 

23 async def get_export_data( 

24 self, 

25 filters: dict[str, Any], 

26 columns: list[str], 

27 sort_by: str | None = None, 

28 sort_order: str = "asc", 

29 limit: int | None = None, 

30 offset: int | None = None, 

31 ) -> list[dict[str, Any]]: 

32 """Implementation for IExportDataSource.get_export_data.""" 

33 qs = QuerySpec() 

34 

35 # Add filters 

36 for field, val in filters.items(): 

37 if field.endswith("__in"): 

38 qs = qs.with_where_in(field[:-4], list(val)) 

39 else: 

40 qs = qs.with_where_eq(field, val) 

41 

42 # Sort 

43 if sort_by: 

44 qs = qs.with_order_by(sort_by, sort_order) 

45 

46 # Pagination 

47 per_page = limit or 1000 # Default large limit for export 

48 page = (offset // per_page + 1) if offset is not None else 1 

49 qs = qs.with_page(page).with_per_page(per_page) 

50 

51 # Select 

52 if columns: 

53 qs = qs.with_select(*columns) 

54 

55 result = await self.data_source.find_many(qs) 

56 return [ 

57 dict(item) if not isinstance(item, dict) else item for item in result.items 

58 ] 

59 

60 async def get_export_count(self, filters: dict[str, Any]) -> int: 

61 """Implementation for IExportDataSource.get_export_count.""" 

62 qs = QuerySpec() 

63 for field, val in filters.items(): 

64 if field.endswith("__in"): 

65 qs = qs.with_where_in(field[:-4], list(val)) 

66 else: 

67 qs = qs.with_where_eq(field, val) 

68 

69 return await self.data_source.count(qs) 

70 

71 async def get_column_definitions(self) -> list[dict[str, Any]]: 

72 """Get column metadata if possible (simplified).""" 

73 # This would ideally introspect the model or schema 

74 return []