Coverage for src/lexigram/admin/services/import_/service.py: 0%
163 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +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 allowed_fields: set[str] | None = None,
256 ) -> None:
257 self._data_source = data_source
258 self._required_fields: list[str] = required_fields or []
259 self._max_rows = max_rows
260 # Mass-assignment guard: when set, rows carrying keys outside this
261 # allowlist are rejected during validation.
262 self._allowed_fields = allowed_fields
263 self._reports: list[ImportReport] = []
265 # ------------------------------------------------------------------
266 # Failed-import reports
267 # ------------------------------------------------------------------
269 def reports(self) -> list[ImportReport]:
270 """Return all stored failed-import reports (most recent last).
272 Returns:
273 Copy of the in-memory report list; empty when no imports
274 have failed.
275 """
276 return list(self._reports)
278 def get_report(self, report_id: str) -> ImportReport | None:
279 """Look up a stored failed-import report by id.
281 Args:
282 report_id: Report identifier from :meth:`reports`.
284 Returns:
285 The matching report, or None when unknown.
286 """
287 for report in self._reports:
288 if report.id == report_id:
289 return report
290 return None
292 def delete_report(self, report_id: str) -> bool:
293 """Remove a stored failed-import report.
295 Args:
296 report_id: Report identifier from :meth:`reports`.
298 Returns:
299 True when the report was removed, False when unknown.
300 """
301 for index, report in enumerate(self._reports):
302 if report.id == report_id:
303 del self._reports[index]
304 return True
305 return False
307 def _store_report(self, job: ImportJob, result: ImportResult) -> None:
308 """Persist a failed-import report for the given commit result."""
309 from lexigram.identity import ambient as identity
310 from lexigram.primitives import clock
312 self._reports.append(
313 ImportReport(
314 id=identity.new_uuid(),
315 source_filename=job.source_filename,
316 created_at=clock.now().isoformat(),
317 total_rows=len(job.rows),
318 failed_rows=result.failed,
319 failures=[*job.errors, *result.errors],
320 )
321 )
323 # ------------------------------------------------------------------
324 # Public API
325 # ------------------------------------------------------------------
327 async def parse(
328 self,
329 content: bytes,
330 filename: str,
331 *,
332 column_map: dict[str, str] | None = None,
333 ) -> Result[ImportJob, AdminError]:
334 """Parse uploaded file bytes and return a validated ImportJob.
336 Validation is non-destructive — nothing is written to the data source.
338 Args:
339 content: Raw file bytes.
340 filename: Original filename; used to detect format (csv / json).
341 column_map: Optional explicit header-to-field mapping.
342 CSV: ``{"CSV Header": "model_field"}``
343 JSON: ``{"json_key": "model_field"}``
344 Defaults to identity mapping (header == field).
346 Returns:
347 ``Result[ImportJob, AdminError]`` — Ok even when rows have
348 validation errors (so callers can show the preview).
349 Err only on file-level failures (wrong format, size limit).
350 """
351 lower = filename.lower()
352 if lower.endswith(".csv"):
353 rows, effective_map, parse_errors = _parse_csv(
354 content, column_map=column_map
355 )
356 elif lower.endswith((".json", ".jsonl")):
357 rows, effective_map, parse_errors = _parse_json(
358 content, column_map=column_map
359 )
360 else:
361 return Err(
362 AdminError(
363 message=f"Unsupported file format: {filename!r}. Use .csv or .json."
364 )
365 )
367 if parse_errors and not rows:
368 return Err(AdminError(message=parse_errors[0].message))
370 if self._max_rows and len(rows) > self._max_rows:
371 return Err(
372 AdminError(
373 message=f"File contains {len(rows):,} rows which exceeds the limit of {self._max_rows:,}."
374 )
375 )
377 # Row-level validation
378 validation_errors = list(parse_errors)
379 validation_errors.extend(self._validate_rows(rows))
381 job = ImportJob(
382 rows=rows,
383 errors=validation_errors,
384 column_map=effective_map,
385 source_filename=filename,
386 total_rows=len(rows),
387 )
388 logger.info(
389 "Import parsed: filename=%s rows=%d errors=%d",
390 filename,
391 len(rows),
392 len(validation_errors),
393 )
394 return Ok(job)
396 async def commit(self, job: ImportJob) -> Result[ImportResult, AdminError]:
397 """Write valid rows from *job* to the data source.
399 Rows with errors are skipped. Remaining rows are inserted one at a
400 time; individual insert failures are collected and do not abort the
401 rest of the batch.
403 Args:
404 job: A parsed ImportJob (from :meth:`parse`).
406 Returns:
407 ``Result[ImportResult, AdminError]`` — Ok with counts and any
408 per-row commit errors.
409 """
410 created = 0
411 failed = 0
412 commit_errors: list[ImportRowError] = []
413 error_row_set = {e.row for e in job.errors}
415 for row_num, row in enumerate(job.rows, start=1):
416 if row_num in error_row_set:
417 failed += 1
418 continue
419 try:
420 await self._data_source.create(row)
421 created += 1
422 except (ValueError, TypeError, KeyError, RuntimeError) as exc:
423 failed += 1
424 commit_errors.append(
425 ImportRowError(row=row_num, field="__row__", message=str(exc))
426 )
427 logger.warning("Import commit error at row %d: %s", row_num, exc)
429 result = ImportResult(created=created, failed=failed, errors=commit_errors)
430 if result.failed:
431 self._store_report(job, result)
432 logger.info(
433 "Import committed: filename=%s created=%d failed=%d",
434 job.source_filename,
435 created,
436 failed,
437 )
438 return Ok(result)
440 # ------------------------------------------------------------------
441 # Internal helpers
442 # ------------------------------------------------------------------
444 def _validate_rows(self, rows: list[dict[str, Any]]) -> list[ImportRowError]:
445 """Run required-field validation over all rows.
447 Args:
448 rows: Parsed rows to validate.
450 Returns:
451 List of ImportRowError for any violations found.
452 """
453 errors: list[ImportRowError] = []
454 for row_num, row in enumerate(rows, start=1):
455 if self._allowed_fields is not None:
456 for key in row:
457 if key not in self._allowed_fields:
458 errors.append(
459 ImportRowError(
460 row=row_num,
461 field=key,
462 message=f"Unknown field '{key}' (not importable)",
463 )
464 )
465 for field_name in self._required_fields:
466 val = row.get(field_name)
467 if val is None or (isinstance(val, str) and not val.strip()):
468 errors.append(
469 ImportRowError(
470 row=row_num,
471 field=field_name,
472 message=f"'{field_name}' is required",
473 )
474 )
475 return errors
478__all__ = [
479 "AdminImportService",
480 "ImportJob",
481 "ImportResult",
482 "ImportRowError",
483]