Coverage for src/lexigram/admin/data/adapters/memory_adapter.py: 0%
134 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""In-memory data source adapter for Lexigram Admin."""
3from __future__ import annotations
5from typing import Any, Generic, TypeVar
7from lexigram.admin.data.data_source import IDataSource, QueryResult
8from lexigram.admin.data.query import FilterOperator, QuerySpec
9from lexigram.di.decorators import inject
11T = TypeVar("T")
14@inject
15class InMemoryDataSource(IDataSource[T], Generic[T]):
16 """Data source for in-memory collections of dictionaries.
18 This is useful for testing, mocking, or displaying static data
19 using the Admin UI components.
20 """
22 returns_result: bool = False # Marker: this adapter returns QueryResult, not Result
24 def __init__(self, data: list[T]) -> None:
25 """Initialize with a list of data items.
27 Args:
28 data: List of dictionary-like objects.
29 """
30 self._items = list(data)
32 async def find_one(self, item_id: Any) -> T | None:
33 """Find a single item by its 'id' field."""
34 for item in self._items:
35 # We assume items are dicts or have an .id attribute
36 item_identifier = (
37 item.get("id") if isinstance(item, dict) else getattr(item, "id", None)
38 )
39 if str(item_identifier) == str(item_id):
40 return item
41 return None
43 async def find_many(self, query: QuerySpec) -> QueryResult[T]:
44 """Filter, sort, and paginate in-memory items."""
45 filtered_items = self._apply_filters(self._items, query)
47 # Search
48 if query.search and query.search_fields:
49 term = query.search.lower()
50 searched = []
51 for item in filtered_items:
52 match = False
53 for field in query.search_fields:
54 val = (
55 item.get(field)
56 if isinstance(item, dict)
57 else getattr(item, field, None)
58 )
59 if val and term in str(val).lower():
60 match = True
61 break
62 if match:
63 searched.append(item)
64 filtered_items = searched
66 # Sort
67 if query.sort_by:
68 field = query.sort_by
69 reverse = query.sort_order == "desc"
71 def get_val(x: Any) -> Any:
72 val = x.get(field) if isinstance(x, dict) else getattr(x, field, None)
73 return "" if val is None else val
75 filtered_items.sort(key=get_val, reverse=reverse)
77 total = len(filtered_items)
79 # Paginate
80 start = (query.page - 1) * query.per_page
81 end = start + query.per_page
82 paginated_items = filtered_items[start:end]
84 return QueryResult(
85 items=paginated_items,
86 total=total,
87 page=query.page,
88 per_page=query.per_page,
89 has_next=end < total,
90 has_prev=query.page > 1,
91 )
93 async def count(self, query: QuerySpec) -> int:
94 """Count items matching the query."""
95 filtered_items = self._apply_filters(self._items, query)
96 return len(filtered_items)
98 async def create(self, data: dict[str, Any]) -> T:
99 """Create a new item in memory."""
100 # Check if it has an id, if not generate one (naive)
101 if "id" not in data:
102 data = dict(data)
103 data["id"] = len(self._items) + 1
105 # In a real generic implementation, we'd need to cast to T
106 new_item = data
107 self._items.append(new_item) # type: ignore[arg-type]
108 return new_item # type: ignore[return-value]
110 async def update(self, item_id: Any, data: dict[str, Any]) -> T:
111 """Update an item in memory."""
112 for i, item in enumerate(self._items):
113 item_identifier = (
114 item.get("id") if isinstance(item, dict) else getattr(item, "id", None)
115 )
116 if str(item_identifier) == str(item_id):
117 if isinstance(item, dict):
118 updated_item: Any = {**item, **data, "id": item_id}
119 self._items[i] = updated_item
120 return self._items[i]
121 for k, v in data.items():
122 setattr(item, k, v)
123 return item
124 raise ValueError(f"Item with id {item_id} not found")
126 async def delete(self, item_id: Any) -> bool:
127 """Delete an item from memory."""
128 initial_len = len(self._items)
129 self._items = [
130 item
131 for item in self._items
132 if str(
133 item.get("id") if isinstance(item, dict) else getattr(item, "id", None),
134 )
135 != str(item_id)
136 ]
137 return len(self._items) < initial_len
139 async def bulk_create(self, items: list[dict[str, Any]]) -> list[T]:
140 """Bulk create items in memory."""
141 results = []
142 for item_data in items:
143 results.append(await self.create(item_data))
144 return results
146 async def bulk_update(self, ids: list[Any], data: dict[str, Any]) -> int:
147 """Bulk update items in memory."""
148 count = 0
149 for id_ in ids:
150 try:
151 await self.update(id_, data)
152 count += 1
153 except ValueError:
154 continue
155 return count
157 async def bulk_delete(self, ids: list[Any]) -> int:
158 """Bulk delete items in memory."""
159 count = 0
160 ids_to_delete = list(map(str, ids))
162 new_items = []
163 for item in self._items:
164 item_id = (
165 item.get("id") if isinstance(item, dict) else getattr(item, "id", None)
166 )
167 if str(item_id) in ids_to_delete:
168 count += 1
169 else:
170 new_items.append(item)
172 self._items = new_items
173 return count
175 def _apply_filters(self, items: list[T], query: QuerySpec) -> list[T]:
176 """Apply all specified filters to the item list."""
177 conditions = query.filter_conditions
178 if not conditions:
179 return list(items)
181 filtered = list(items)
182 for condition in conditions:
183 field = condition.field
184 op_type = condition.operator
185 val = condition.value
187 # Map operator
188 filtered = [
189 item for item in filtered if self._matches(item, field, op_type, val)
190 ]
192 return filtered
194 def _matches(
195 self,
196 item: Any,
197 field: str,
198 op_type: FilterOperator,
199 target_val: Any,
200 ) -> bool:
201 """Check if an item matches a specific filter condition."""
202 # Get item value
203 item_val = (
204 item.get(field) if isinstance(item, dict) else getattr(item, field, None)
205 )
207 if op_type == FilterOperator.EQ:
208 return item_val == target_val
209 if op_type == FilterOperator.NEQ:
210 return item_val != target_val
211 if op_type == FilterOperator.GT:
212 return item_val > target_val
213 if op_type == FilterOperator.GTE:
214 return item_val >= target_val
215 if op_type == FilterOperator.LT:
216 return item_val < target_val
217 if op_type == FilterOperator.LTE:
218 return item_val <= target_val
219 if op_type == FilterOperator.IN:
220 return item_val in target_val
221 if op_type == FilterOperator.NOT_IN:
222 return item_val not in target_val
223 if op_type == FilterOperator.CONTAINS:
224 return target_val.lower() in str(item_val).lower()
225 if op_type == FilterOperator.ICONTAINS:
226 return target_val.lower() in str(item_val).lower()
227 if op_type == FilterOperator.IS_NULL:
228 return item_val is None
229 if op_type == FilterOperator.BETWEEN:
230 min_v, max_v = target_val
231 return min_v <= item_val <= max_v
233 return False