Coverage for src/lexigram/admin/data/data_source.py: 86%
104 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Unified data source interfaces for Lexigram Admin."""
3from __future__ import annotations
5from dataclasses import dataclass
6from typing import Any, Generic, Protocol, TypeVar, runtime_checkable
8from lexigram.admin.data.query import QuerySpec
10# Type variable for the entity type
11T = TypeVar("T")
14@dataclass
15class QueryResult(Generic[T]):
16 """Result of a data query with pagination and metadata.
18 Attributes:
19 items: The list of entities returned by the query.
20 total: Total number of entities matching the query (for pagination).
21 page: Current page number (1-indexed).
22 per_page: Number of items per page.
23 has_next: Whether there is a next page.
24 has_prev: Whether there is a previous page.
25 cursor: Optional cursor for cursor-based pagination.
26 """
28 items: list[T]
29 total: int = 0
30 page: int = 1
31 per_page: int = 20
32 has_next: bool = False
33 has_prev: bool = False
34 cursor: str | None = None
37@runtime_checkable
38class IDataSource(Protocol[T]):
39 """Protocol for unified data access.
41 .. stability:: stable
43 This interface abstracts away the underlying data storage (DB, API, etc.)
44 providing a consistent API for the Admin UI and Resource Managers.
45 """
47 async def find_one(self, item_id: Any) -> T | None:
48 """Find a single entity by its unique identifier.
50 Args:
51 item_id: The unique identifier of the entity.
53 Returns:
54 The entity if found, None otherwise.
55 """
56 ...
58 async def find_many(self, query: QuerySpec) -> QueryResult[T]:
59 """Find multiple entities matching the given query specification.
61 Args:
62 query: The query specification (typically a QuerySpec).
64 Returns:
65 A QueryResult containing the requested entities and metadata.
66 """
67 ...
69 async def count(self, query: QuerySpec) -> int:
70 """Count the total number of entities matching the given query.
72 Args:
73 query: The query specification.
75 Returns:
76 The total count of matching entities.
77 """
78 ...
80 async def create(self, data: dict[str, Any]) -> T:
81 """Create a new entity with the provided data.
83 Args:
84 data: Dictionary of entity attributes.
86 Returns:
87 The created entity.
88 """
89 ...
91 async def update(self, item_id: Any, data: dict[str, Any]) -> T:
92 """Update an existing entity.
94 Args:
95 item_id: The unique identifier of the entity to update.
96 data: Dictionary of attributes to update.
98 Returns:
99 The updated entity.
100 """
101 ...
103 async def delete(self, item_id: Any) -> bool:
104 """Delete an entity by its unique identifier.
106 Args:
107 item_id: The unique identifier of the entity.
109 Returns:
110 True if deleted successfully, False otherwise.
111 """
112 ...
114 async def bulk_create(self, items: list[dict[str, Any]]) -> list[T]:
115 """Create multiple entities in a single operation.
117 Args:
118 items: List of dictionaries containing entity data.
120 Returns:
121 List of created entities.
122 """
123 ...
125 async def bulk_update(self, ids: list[Any], data: dict[str, Any]) -> int:
126 """Update multiple entities matching the provided IDs with the same data.
128 Args:
129 ids: List of entity identifiers.
130 data: Dictionary of attributes to update across all entities.
132 Returns:
133 The count of updated entities.
134 """
135 ...
137 async def bulk_delete(self, ids: list[Any]) -> int:
138 """Delete multiple entities by their identifiers.
140 Args:
141 ids: List of entity identifiers.
143 Returns:
144 The count of deleted entities.
145 """
146 ...
149# ---------------------------------------------------------------------------
150# Concrete base classes (moved from interfaces/data_source.py)
151# ---------------------------------------------------------------------------
153from abc import ABC, abstractmethod
154from typing import TYPE_CHECKING
156if TYPE_CHECKING:
157 from lexigram.contracts import (
158 DatabaseProviderProtocol,
159 )
162def _quote_identifier(name: str) -> str:
163 """Quote SQL identifier to prevent injection.
165 Simple identifier validation - allows alphanumeric and underscore only.
166 """
167 if not name or not name.replace("_", "").isalnum():
168 raise ValueError(f"Invalid SQL identifier: {name}")
169 return f'"{name}"'
172class DataSourceBase(ABC, Generic[T]):
173 """Abstract base class for data sources.
175 Provides a consistent interface for CRUD operations
176 that can be implemented by different backends (SQL, API, memory, etc.).
177 """
179 @abstractmethod
180 async def find_one(self, item_id: Any) -> T | None:
181 """Find a single entity by ID."""
182 ...
184 @abstractmethod
185 async def update(
186 self,
187 item_id: Any,
188 data: dict[str, Any] | None = None,
189 ) -> T | None:
190 """Update an existing entity."""
191 ...
193 @abstractmethod
194 async def delete(self, item_id: Any) -> bool:
195 """Delete an entity by ID."""
196 ...
198 @abstractmethod
199 async def find_many(
200 self, query: QuerySpec | None = None, **filters: Any
201 ) -> QueryResult[T]:
202 """Find multiple entities with optional filtering."""
203 ...
205 @abstractmethod
206 async def create(self, entity: T | dict[str, Any]) -> T:
207 """Create a new entity."""
208 ...
210 async def count(self, query: QuerySpec | None = None) -> int:
211 """Count entities matching query."""
212 result = await self.find_many(query)
213 return result.total
216class SqlDataSource(DataSourceBase[T]):
217 """SQL-based data source using lexigram-sql.
219 Provides a base class for services that need SQL database access.
220 Subclasses should implement domain-specific query methods.
221 """
223 def __init__(
224 self,
225 db: DatabaseProviderProtocol,
226 table_name: str,
227 *,
228 id_field: str = "id",
229 ):
230 self.db = db
231 self.table_name = table_name
232 self.id_field = id_field
234 async def find_one(self, item_id: Any) -> T | None:
235 """Find a single entity by ID."""
236 table = _quote_identifier(self.table_name)
237 id_field = _quote_identifier(self.id_field)
238 query = f"SELECT * FROM {table} WHERE {id_field} = $1" # noqa: S608 — table/id_field pre-validated by _quote_identifier (alnum+underscore only)
239 return await self.db.fetch_one(query, [item_id]) # type: ignore[attr-defined]
241 async def find_many(
242 self,
243 query: QuerySpec | None = None,
244 *,
245 page: int = 1,
246 per_page: int = 20,
247 **filters: Any,
248 ) -> QueryResult[T]:
249 """Find multiple entities with pagination."""
250 table = _quote_identifier(self.table_name)
251 base_query = f"SELECT * FROM {table}" # noqa: S608 — table pre-validated by _quote_identifier (alnum+underscore only)
252 count_query = f"SELECT COUNT(*) FROM {table}" # noqa: S608 — table pre-validated by _quote_identifier (alnum+underscore only)
254 where_clauses: list[str] = []
255 params: list[Any] = []
256 for i, (field, value) in enumerate(filters.items(), 1):
257 where_clauses.append(f"{_quote_identifier(field)} = ${i}")
258 params.append(value)
260 if where_clauses:
261 where_sql = " WHERE " + " AND ".join(where_clauses)
262 base_query += where_sql
263 count_query += where_sql
265 offset = (page - 1) * per_page
266 base_query += f" LIMIT {per_page} OFFSET {offset}"
268 items = await self.db.fetch_all(base_query, params) # type: ignore[attr-defined]
269 count_result = await self.db.fetch_one(count_query, params) # type: ignore[attr-defined]
270 total = count_result[0] if count_result else 0
272 return QueryResult(
273 items=list(items),
274 total=total,
275 page=page,
276 per_page=per_page,
277 has_next=offset + per_page < total,
278 has_prev=page > 1,
279 )
281 async def create(self, entity: T | dict[str, Any]) -> T:
282 """Create a new entity."""
283 data = (
284 entity
285 if isinstance(entity, dict)
286 else (
287 entity.model_dump()
288 if hasattr(entity, "model_dump")
289 else entity.__dict__
290 )
291 )
292 fields = list(data.keys())
293 placeholders = [f"${i}" for i in range(1, len(fields) + 1)]
294 values = list(data.values())
295 query = f"""
296 INSERT INTO {_quote_identifier(self.table_name)} ({", ".join(_quote_identifier(f) for f in fields)})
297 VALUES ({", ".join(placeholders)})
298 RETURNING *
299 """ # noqa: S608 — identifiers pre-validated by _quote_identifier (alnum+underscore only)
300 return await self.db.fetch_one(query, values) # type: ignore[attr-defined]
302 async def update(
303 self,
304 item_id: Any,
305 data: dict[str, Any] | None = None,
306 ) -> T | None:
307 """Update an existing entity."""
308 if data is None:
309 entity = item_id
310 data = (
311 entity
312 if isinstance(entity, dict)
313 else (
314 entity.model_dump()
315 if hasattr(entity, "model_dump")
316 else entity.__dict__
317 )
318 )
319 item_id = data.get(self.id_field)
320 if not data:
321 return None
322 update_data = {k: v for k, v in data.items() if k != self.id_field}
323 set_clauses = [
324 f"{_quote_identifier(field)} = ${i}"
325 for i, field in enumerate(update_data.keys(), 1)
326 ]
327 values = list(update_data.values())
328 values.append(item_id)
329 query = f"""
330 UPDATE {_quote_identifier(self.table_name)}
331 SET {", ".join(set_clauses)}
332 WHERE {_quote_identifier(self.id_field)} = ${len(values)}
333 RETURNING *
334 """ # noqa: S608 — identifiers pre-validated by _quote_identifier (alnum+underscore only)
335 return await self.db.fetch_one(query, values) # type: ignore[attr-defined]
337 async def delete(self, item_id: Any) -> bool:
338 """Delete an entity by ID."""
339 if hasattr(item_id, self.id_field):
340 item_id = getattr(item_id, self.id_field)
341 elif isinstance(item_id, dict):
342 item_id = item_id.get(self.id_field, item_id)
343 table = _quote_identifier(self.table_name)
344 id_field = _quote_identifier(self.id_field)
345 query = f"DELETE FROM {table} WHERE {id_field} = $1" # noqa: S608 — table/id_field pre-validated by _quote_identifier (alnum+underscore only)
346 result = await self.db.execute(query, [item_id])
347 return result > 0 if isinstance(result, int) else True