Coverage for src/lexigram/admin/services/saved_filters.py: 0%
97 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Saved filter service for lexigram-admin.
3Allows users to save named filter combinations per resource and quickly
4re-apply them from a dropdown in the filter panel.
6Filter sets are stored as JSON blobs in the admin_saved_filters table (or
7in-memory when no database is configured).
9Usage::
11 svc = SavedFiltersService()
12 await svc.save("users", "Active Admins", {"role": "admin", "is_active": "1"}, user_id="u1")
13 sets = await svc.list_for_resource("users", user_id="u1")
14 await svc.delete("users", "Active Admins", user_id="u1")
15"""
17from __future__ import annotations
19from dataclasses import dataclass, field
20from datetime import UTC, datetime
21from typing import Any, Protocol, runtime_checkable
23from lexigram import serialization as json
24from lexigram.admin.exceptions import AdminError, NotFoundError
25from lexigram.di.decorators import inject
26from lexigram.logging import get_logger
27from lexigram.result import Err, Ok, Result
29logger = get_logger(__name__)
32# ---------------------------------------------------------------------------
33# Data structures
34# ---------------------------------------------------------------------------
37@dataclass(frozen=True)
38class SavedFilterSet:
39 """A named filter combination saved by a user for a resource.
41 Attributes:
42 resource: Resource name (e.g. ``"users"``).
43 name: Human-readable label for this filter set.
44 filters: Mapping of field name → filter value.
45 user_id: Owner of this saved filter (``None`` = shared / global).
46 created_at: When the filter was saved.
47 """
49 resource: str
50 name: str
51 filters: dict[str, Any]
52 user_id: str | None = None
53 created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
56# ---------------------------------------------------------------------------
57# Storage protocol (structural)
58# ---------------------------------------------------------------------------
61@runtime_checkable
62class _FilterRepository(Protocol):
63 """Narrow structural protocol for the saved-filters storage backend."""
65 async def find_many(
66 self, *, limit: int = 100, offset: int = 0, **filters: Any
67 ) -> list[Any]: ...
69 async def create(self, data: Any) -> Any: ...
71 async def delete(self, item_id: Any) -> bool: ...
73 async def find_one(self, **filters: Any) -> Any | None: ...
76# ---------------------------------------------------------------------------
77# In-memory fallback backend
78# ---------------------------------------------------------------------------
81class _InMemoryFilterStore:
82 """Ephemeral in-memory store used when no database backend is provided."""
84 def __init__(self) -> None:
85 self._records: list[dict[str, Any]] = []
86 self._next_id: int = 1
88 async def find_many(
89 self, *, limit: int = 100, offset: int = 0, **filters: Any
90 ) -> list[Any]:
91 results = []
92 for rec in self._records:
93 if all(rec.get(k) == v for k, v in filters.items()):
94 results.append(rec)
95 return results[offset : offset + limit]
97 async def create(self, data: Any) -> Any:
98 record = dict(data) if not isinstance(data, dict) else data
99 record["id"] = self._next_id
100 self._next_id += 1
101 self._records.append(record)
102 return record
104 async def delete(self, item_id: Any) -> bool:
105 before = len(self._records)
106 self._records = [r for r in self._records if r.get("id") != item_id]
107 return len(self._records) < before
109 async def find_one(self, **filters: Any) -> Any | None:
110 for rec in self._records:
111 if all(rec.get(k) == v for k, v in filters.items()):
112 return rec
113 return None
116# ---------------------------------------------------------------------------
117# Service
118# ---------------------------------------------------------------------------
121@inject
122class SavedFiltersService:
123 """Manages named filter presets per resource and user.
125 Args:
126 repository: Optional storage backend implementing :class:`_FilterRepository`.
127 Falls back to an ephemeral in-memory store when ``None``.
128 max_per_user_per_resource: Maximum saved filter sets allowed per
129 user × resource combination (0 = unlimited).
130 """
132 def __init__(
133 self,
134 repository: _FilterRepository | None = None,
135 *,
136 max_per_user_per_resource: int = 20,
137 ) -> None:
138 self._repo: _FilterRepository = repository or _InMemoryFilterStore()
139 self._max = max_per_user_per_resource
141 # ------------------------------------------------------------------
142 # Public API
143 # ------------------------------------------------------------------
145 async def save(
146 self,
147 resource: str,
148 name: str,
149 filters: dict[str, Any],
150 *,
151 user_id: str | None = None,
152 ) -> Result[SavedFilterSet, AdminError]:
153 """Save a named filter set.
155 If a filter set with the same ``resource``+``name``+``user_id`` already
156 exists it is overwritten.
158 Args:
159 resource: Resource name (e.g. ``"users"``).
160 name: Label for the filter set.
161 filters: Field → value mapping to store.
162 user_id: User who owns this preset (``None`` = global preset).
164 Returns:
165 ``Result[SavedFilterSet, AdminError]``.
166 """
167 if not name.strip():
168 return Err(AdminError(message="Filter set name must not be empty"))
169 if not filters:
170 return Err(
171 AdminError(message="Filter set must contain at least one filter")
172 )
174 # Enforce per-user-per-resource limit
175 if self._max:
176 existing = await self._repo.find_many(resource=resource, user_id=user_id)
177 if len(existing) >= self._max:
178 # Check if this name already exists (overwrite is fine)
179 match = next((r for r in existing if r.get("name") == name), None)
180 if match is None:
181 return Err(
182 AdminError(
183 message=f"Maximum of {self._max} saved filters reached for this resource"
184 )
185 )
187 # Upsert: delete existing with same key before inserting
188 existing_row = await self._repo.find_one(
189 resource=resource, name=name, user_id=user_id
190 )
191 if existing_row is not None:
192 row_id = existing_row.get("id")
193 if row_id is not None:
194 await self._repo.delete(row_id)
196 now = datetime.now(UTC)
197 record = {
198 "resource": resource,
199 "name": name,
200 "filters_json": json.dumps(filters).decode()
201 if isinstance(json.dumps(filters), bytes)
202 else json.dumps(filters),
203 "user_id": user_id,
204 "created_at": now.isoformat(),
205 }
206 await self._repo.create(record)
208 saved = SavedFilterSet(
209 resource=resource,
210 name=name,
211 filters=filters,
212 user_id=user_id,
213 created_at=now,
214 )
215 logger.info(
216 "Saved filter set: resource=%s name=%r user_id=%s", resource, name, user_id
217 )
218 return Ok(saved)
220 async def list_for_resource(
221 self,
222 resource: str,
223 *,
224 user_id: str | None = None,
225 ) -> list[SavedFilterSet]:
226 """List all saved filter sets for a resource.
228 Returns both the user's own presets and global (user_id=None) presets.
230 Args:
231 resource: Resource name.
232 user_id: Return presets owned by this user plus global ones.
234 Returns:
235 List of :class:`SavedFilterSet` sorted by name.
236 """
237 rows: list[Any] = []
239 # User's own presets
240 if user_id is not None:
241 rows.extend(await self._repo.find_many(resource=resource, user_id=user_id))
243 # Global presets (shared)
244 global_rows = await self._repo.find_many(resource=resource, user_id=None)
245 rows.extend(global_rows)
247 result: list[SavedFilterSet] = []
248 for row in rows:
249 try:
250 filters_raw = row.get("filters_json", "{}")
251 filters = (
252 json.loads(filters_raw)
253 if isinstance(filters_raw, (str, bytes))
254 else {}
255 )
256 created_at_raw = row.get("created_at")
257 created_at = (
258 datetime.fromisoformat(created_at_raw)
259 if isinstance(created_at_raw, str)
260 else datetime.now(UTC)
261 )
262 result.append(
263 SavedFilterSet(
264 resource=row.get("resource", resource),
265 name=row.get("name", ""),
266 filters=filters,
267 user_id=row.get("user_id"),
268 created_at=created_at,
269 )
270 )
271 except (KeyError, ValueError, TypeError):
272 logger.warning("Skipping malformed saved filter row: %s", row)
274 return sorted(result, key=lambda s: s.name)
276 async def delete(
277 self,
278 resource: str,
279 name: str,
280 *,
281 user_id: str | None = None,
282 ) -> Result[None, NotFoundError]:
283 """Delete a saved filter set by resource + name + owner.
285 Args:
286 resource: Resource name.
287 name: Filter set label.
288 user_id: Owner user ID.
290 Returns:
291 ``Result[None, NotFoundError]``.
292 """
293 row = await self._repo.find_one(resource=resource, name=name, user_id=user_id)
294 if row is None:
295 return Err(
296 NotFoundError(
297 message=f"Saved filter '{name}' not found for resource '{resource}'"
298 )
299 )
301 row_id = row.get("id")
302 if row_id is not None:
303 await self._repo.delete(row_id)
305 return Ok(None)
308__all__ = [
309 "SavedFilterSet",
310 "SavedFiltersService",
311]