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

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 qs = qs.with_where_eq(field, val) 

38 

39 # Sort 

40 if sort_by: 

41 qs = qs.with_order_by(sort_by, sort_order) 

42 

43 # Pagination 

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

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

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

47 

48 # Select 

49 if columns: 

50 qs = qs.with_select(*columns) 

51 

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

53 return [ 

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

55 ] 

56 

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

58 """Implementation for IExportDataSource.get_export_count.""" 

59 qs = QuerySpec() 

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

61 qs = qs.with_where_eq(field, val) 

62 

63 return await self.data_source.count(qs) 

64 

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

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

67 # This would ideally introspect the model or schema 

68 return []