Coverage for src/lexigram/admin/services/import_/service.py: 99%
158 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Bulk import service for lexigram-admin.
3Supports CSV and JSON file uploads with:
4- Column mapping from file headers to resource fields
5- Row-level validation before commit (preview mode)
6- Structured error reporting per row
7- Streaming parse to keep memory usage bounded
9Usage::
11 service = AdminImportService(data_source=ds)
12 job = await service.parse(raw_bytes, filename="users.csv")
13 if job.has_errors:
14 return job # show preview with errors
15 result = await service.commit(job)
16"""
18from __future__ import annotations
20import csv
21from dataclasses import dataclass, field
22import io
23from typing import Any
25from lexigram import serialization as json
26from lexigram.admin.exceptions import AdminError
27from lexigram.di.decorators import inject
28from lexigram.logging import get_logger
29from lexigram.result import Err, Ok, Result
31logger = get_logger(__name__)
34# ---------------------------------------------------------------------------
35# Data structures
36# ---------------------------------------------------------------------------
39@dataclass
40class ImportRowError:
41 """Validation error for a single import row.
43 Attributes:
44 row: 1-indexed row number within the file.
45 field: Field name that failed validation, or ``"__row__"`` for row-level errors.
46 message: Human-readable error description.
47 """
49 row: int
50 field: str
51 message: str
54@dataclass
55class ImportJob:
56 """Parsed import batch ready for validation or commit.
58 Attributes:
59 rows: Parsed rows as dicts keyed by mapped field names.
60 errors: Per-row validation errors collected during :meth:`AdminImportService.parse`.
61 column_map: Mapping from source file header → target resource field name.
62 source_filename: Original uploaded filename.
63 total_rows: Total number of data rows (excludes header).
64 """
66 rows: list[dict[str, Any]]
67 errors: list[ImportRowError]
68 column_map: dict[str, str]
69 source_filename: str
70 total_rows: int
72 @property
73 def has_errors(self) -> bool:
74 """True when at least one validation error was found during parse."""
75 return bool(self.errors)
77 @property
78 def valid_rows(self) -> list[dict[str, Any]]:
79 """Rows that had no validation errors."""
80 error_rows = {e.row for e in self.errors}
81 return [r for i, r in enumerate(self.rows, start=1) if i not in error_rows]
84@dataclass
85class ImportResult:
86 """Summary returned after a committed import.
88 Attributes:
89 created: Number of records successfully inserted.
90 failed: Number of records that failed during insert.
91 errors: Errors encountered during commit (row-level).
92 """
94 created: int
95 failed: int
96 errors: list[ImportRowError] = field(default_factory=list)
98 @property
99 def total(self) -> int:
100 """Total records attempted."""
101 return self.created + self.failed
104@dataclass
105class ImportReport:
106 """Stored failed-import report for post-hoc download.
108 Persisted by :class:`AdminImportService` whenever a commit has
109 failed rows, so callers can surface a downloadable error report
110 (mirrors Filament's "Reviewing failed rows" flow).
112 Attributes:
113 id: Unique report identifier.
114 source_filename: Original uploaded filename.
115 created_at: ISO-8601 timestamp of report creation.
116 total_rows: Number of data rows attempted.
117 failed_rows: Number of rows that failed.
118 failures: Per-row validation/commit errors.
119 """
121 id: str
122 source_filename: str
123 created_at: str
124 total_rows: int
125 failed_rows: int
126 failures: list[ImportRowError]
128 def to_csv(self) -> str:
129 """Serialize failed rows as CSV (row number, field, message)."""
130 buffer = io.StringIO()
131 writer = csv.writer(buffer, lineterminator="\n")
132 writer.writerow(["row", "field", "message"])
133 for err in self.failures:
134 writer.writerow([err.row, err.field, err.message])
135 return buffer.getvalue()
138# ---------------------------------------------------------------------------
139# Parsers
140# ---------------------------------------------------------------------------
143def _parse_csv(
144 content: bytes,
145 *,
146 column_map: dict[str, str] | None = None,
147) -> tuple[list[dict[str, Any]], dict[str, str], list[ImportRowError]]:
148 """Parse CSV bytes into rows and infer column_map if not provided.
150 Args:
151 content: Raw file bytes.
152 column_map: Optional explicit header → field mapping.
153 If None, headers are used as-is.
155 Returns:
156 Tuple of (rows, effective_column_map, parse_errors).
157 """
158 text = content.decode("utf-8-sig", errors="replace")
159 reader = csv.DictReader(io.StringIO(text))
160 headers: list[str] = list(reader.fieldnames) if reader.fieldnames else []
162 effective_map: dict[str, str] = dict(column_map or {h: h for h in headers})
164 rows: list[dict[str, Any]] = []
165 errors: list[ImportRowError] = []
167 for _row_num, raw_row in enumerate(reader, start=1):
168 mapped: dict[str, Any] = {}
169 for src, dst in effective_map.items():
170 val = raw_row.get(src, "").strip()
171 mapped[dst] = val or None
172 rows.append(mapped)
174 return rows, effective_map, errors
177def _parse_json(
178 content: bytes,
179 *,
180 column_map: dict[str, str] | None = None,
181) -> tuple[list[dict[str, Any]], dict[str, str], list[ImportRowError]]:
182 """Parse JSON bytes (array of objects) into rows.
184 Args:
185 content: Raw file bytes containing a JSON array.
186 column_map: Optional key remapping (source_key → target_field).
188 Returns:
189 Tuple of (rows, effective_column_map, parse_errors).
190 """
191 errors: list[ImportRowError] = []
192 try:
193 data = json.loads(content.decode("utf-8-sig", errors="replace"))
194 except json.JSONDecodeError as exc:
195 errors.append(
196 ImportRowError(row=0, field="__file__", message=f"Invalid JSON: {exc}")
197 )
198 return [], {}, errors
200 if not isinstance(data, list):
201 errors.append(
202 ImportRowError(
203 row=0, field="__file__", message="JSON root must be an array of objects"
204 )
205 )
206 return [], {}, errors
208 effective_map: dict[str, str] = column_map or {}
209 rows: list[dict[str, Any]] = []
211 for row_num, item in enumerate(data, start=1):
212 if not isinstance(item, dict):
213 errors.append(
214 ImportRowError(
215 row=row_num, field="__row__", message="Expected a JSON object"
216 )
217 )
218 continue
219 if effective_map:
220 mapped: dict[str, Any] = {
221 dst: item.get(src) for src, dst in effective_map.items()
222 }
223 else:
224 mapped = dict(item)
225 rows.append(mapped)
227 return rows, effective_map, errors
230# ---------------------------------------------------------------------------
231# Service
232# ---------------------------------------------------------------------------
235@inject
236class AdminImportService:
237 """Service that parses, validates, and commits bulk imports.
239 Supports CSV and JSON files. Validation is performed during
240 :meth:`parse` so callers can show a preview before committing.
242 Args:
243 data_source: An IDataSource-compatible instance that the
244 committed rows will be written to via ``create()``.
245 required_fields: Field names that must be non-empty on every row.
246 max_rows: Maximum number of rows allowed per import (0 = unlimited).
247 """
249 def __init__(
250 self,
251 data_source: Any,
252 *,
253 required_fields: list[str] | None = None,
254 max_rows: int = 10_000,
255 ) -> None:
256 self._data_source = data_source
257 self._required_fields: list[str] = required_fields or []
258 self._max_rows = max_rows
259 self._reports: list[ImportReport] = []
261 # ------------------------------------------------------------------
262 # Failed-import reports
263 # ------------------------------------------------------------------
265 def reports(self) -> list[ImportReport]:
266 """Return all stored failed-import reports (most recent last).
268 Returns:
269 Copy of the in-memory report list; empty when no imports
270 have failed.
271 """
272 return list(self._reports)
274 def get_report(self, report_id: str) -> ImportReport | None:
275 """Look up a stored failed-import report by id.
277 Args:
278 report_id: Report identifier from :meth:`reports`.
280 Returns:
281 The matching report, or None when unknown.
282 """
283 for report in self._reports:
284 if report.id == report_id:
285 return report
286 return None
288 def delete_report(self, report_id: str) -> bool:
289 """Remove a stored failed-import report.
291 Args:
292 report_id: Report identifier from :meth:`reports`.
294 Returns:
295 True when the report was removed, False when unknown.
296 """
297 for index, report in enumerate(self._reports):
298 if report.id == report_id:
299 del self._reports[index]
300 return True
301 return False
303 def _store_report(self, job: ImportJob, result: ImportResult) -> None:
304 """Persist a failed-import report for the given commit result."""
305 from lexigram.identity import ambient as identity
306 from lexigram.primitives import clock
308 self._reports.append(
309 ImportReport(
310 id=identity.new_uuid(),
311 source_filename=job.source_filename,
312 created_at=clock.now().isoformat(),
313 total_rows=len(job.rows),
314 failed_rows=result.failed,
315 failures=[*job.errors, *result.errors],
316 )
317 )
319 # ------------------------------------------------------------------
320 # Public API
321 # ------------------------------------------------------------------
323 async def parse(
324 self,
325 content: bytes,
326 filename: str,
327 *,
328 column_map: dict[str, str] | None = None,
329 ) -> Result[ImportJob, AdminError]:
330 """Parse uploaded file bytes and return a validated ImportJob.
332 Validation is non-destructive — nothing is written to the data source.
334 Args:
335 content: Raw file bytes.
336 filename: Original filename; used to detect format (csv / json).
337 column_map: Optional explicit header-to-field mapping.
338 CSV: ``{"CSV Header": "model_field"}``
339 JSON: ``{"json_key": "model_field"}``
340 Defaults to identity mapping (header == field).
342 Returns:
343 ``Result[ImportJob, AdminError]`` — Ok even when rows have
344 validation errors (so callers can show the preview).
345 Err only on file-level failures (wrong format, size limit).
346 """
347 lower = filename.lower()
348 if lower.endswith(".csv"):
349 rows, effective_map, parse_errors = _parse_csv(
350 content, column_map=column_map
351 )
352 elif lower.endswith((".json", ".jsonl")):
353 rows, effective_map, parse_errors = _parse_json(
354 content, column_map=column_map
355 )
356 else:
357 return Err(
358 AdminError(
359 message=f"Unsupported file format: {filename!r}. Use .csv or .json."
360 )
361 )
363 if parse_errors and not rows:
364 return Err(AdminError(message=parse_errors[0].message))
366 if self._max_rows and len(rows) > self._max_rows:
367 return Err(
368 AdminError(
369 message=f"File contains {len(rows):,} rows which exceeds the limit of {self._max_rows:,}."
370 )
371 )
373 # Row-level validation
374 validation_errors = list(parse_errors)
375 validation_errors.extend(self._validate_rows(rows))
377 job = ImportJob(
378 rows=rows,
379 errors=validation_errors,
380 column_map=effective_map,
381 source_filename=filename,
382 total_rows=len(rows),
383 )
384 logger.info(
385 "Import parsed: filename=%s rows=%d errors=%d",
386 filename,
387 len(rows),
388 len(validation_errors),
389 )
390 return Ok(job)
392 async def commit(self, job: ImportJob) -> Result[ImportResult, AdminError]:
393 """Write valid rows from *job* to the data source.
395 Rows with errors are skipped. Remaining rows are inserted one at a
396 time; individual insert failures are collected and do not abort the
397 rest of the batch.
399 Args:
400 job: A parsed ImportJob (from :meth:`parse`).
402 Returns:
403 ``Result[ImportResult, AdminError]`` — Ok with counts and any
404 per-row commit errors.
405 """
406 created = 0
407 failed = 0
408 commit_errors: list[ImportRowError] = []
409 error_row_set = {e.row for e in job.errors}
411 for row_num, row in enumerate(job.rows, start=1):
412 if row_num in error_row_set:
413 failed += 1
414 continue
415 try:
416 await self._data_source.create(row)
417 created += 1
418 except (ValueError, TypeError, KeyError, RuntimeError) as exc:
419 failed += 1
420 commit_errors.append(
421 ImportRowError(row=row_num, field="__row__", message=str(exc))
422 )
423 logger.warning("Import commit error at row %d: %s", row_num, exc)
425 result = ImportResult(created=created, failed=failed, errors=commit_errors)
426 if result.failed:
427 self._store_report(job, result)
428 logger.info(
429 "Import committed: filename=%s created=%d failed=%d",
430 job.source_filename,
431 created,
432 failed,
433 )
434 return Ok(result)
436 # ------------------------------------------------------------------
437 # Internal helpers
438 # ------------------------------------------------------------------
440 def _validate_rows(self, rows: list[dict[str, Any]]) -> list[ImportRowError]:
441 """Run required-field validation over all rows.
443 Args:
444 rows: Parsed rows to validate.
446 Returns:
447 List of ImportRowError for any violations found.
448 """
449 errors: list[ImportRowError] = []
450 for row_num, row in enumerate(rows, start=1):
451 for field_name in self._required_fields:
452 val = row.get(field_name)
453 if val is None or (isinstance(val, str) and not val.strip()):
454 errors.append(
455 ImportRowError(
456 row=row_num,
457 field=field_name,
458 message=f"'{field_name}' is required",
459 )
460 )
461 return errors
464__all__ = [
465 "AdminImportService",
466 "ImportJob",
467 "ImportResult",
468 "ImportRowError",
469]