Coverage for src / lexigram / admin / services / import_ / service.py: 0%
123 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +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# ---------------------------------------------------------------------------
105# Parsers
106# ---------------------------------------------------------------------------
109def _parse_csv(
110 content: bytes,
111 *,
112 column_map: dict[str, str] | None = None,
113) -> tuple[list[dict[str, Any]], dict[str, str], list[ImportRowError]]:
114 """Parse CSV bytes into rows and infer column_map if not provided.
116 Args:
117 content: Raw file bytes.
118 column_map: Optional explicit header → field mapping.
119 If None, headers are used as-is.
121 Returns:
122 Tuple of (rows, effective_column_map, parse_errors).
123 """
124 text = content.decode("utf-8-sig", errors="replace")
125 reader = csv.DictReader(io.StringIO(text))
126 headers: list[str] = list(reader.fieldnames) if reader.fieldnames else []
128 effective_map: dict[str, str] = dict(column_map or {h: h for h in headers})
130 rows: list[dict[str, Any]] = []
131 errors: list[ImportRowError] = []
133 for _row_num, raw_row in enumerate(reader, start=1):
134 mapped: dict[str, Any] = {}
135 for src, dst in effective_map.items():
136 val = raw_row.get(src, "").strip()
137 mapped[dst] = val or None
138 rows.append(mapped)
140 return rows, effective_map, errors
143def _parse_json(
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 JSON bytes (array of objects) into rows.
150 Args:
151 content: Raw file bytes containing a JSON array.
152 column_map: Optional key remapping (source_key → target_field).
154 Returns:
155 Tuple of (rows, effective_column_map, parse_errors).
156 """
157 errors: list[ImportRowError] = []
158 try:
159 data = json.loads(content.decode("utf-8-sig", errors="replace"))
160 except json.JSONDecodeError as exc:
161 errors.append(
162 ImportRowError(row=0, field="__file__", message=f"Invalid JSON: {exc}")
163 )
164 return [], {}, errors
166 if not isinstance(data, list):
167 errors.append(
168 ImportRowError(
169 row=0, field="__file__", message="JSON root must be an array of objects"
170 )
171 )
172 return [], {}, errors
174 effective_map: dict[str, str] = column_map or {}
175 rows: list[dict[str, Any]] = []
177 for row_num, item in enumerate(data, start=1):
178 if not isinstance(item, dict):
179 errors.append(
180 ImportRowError(
181 row=row_num, field="__row__", message="Expected a JSON object"
182 )
183 )
184 continue
185 if effective_map:
186 mapped: dict[str, Any] = {
187 dst: item.get(src) for src, dst in effective_map.items()
188 }
189 else:
190 mapped = dict(item)
191 rows.append(mapped)
193 return rows, effective_map, errors
196# ---------------------------------------------------------------------------
197# Service
198# ---------------------------------------------------------------------------
201@inject
202class AdminImportService:
203 """Service that parses, validates, and commits bulk imports.
205 Supports CSV and JSON files. Validation is performed during
206 :meth:`parse` so callers can show a preview before committing.
208 Args:
209 data_source: An IDataSource-compatible instance that the
210 committed rows will be written to via ``create()``.
211 required_fields: Field names that must be non-empty on every row.
212 max_rows: Maximum number of rows allowed per import (0 = unlimited).
213 """
215 def __init__(
216 self,
217 data_source: Any,
218 *,
219 required_fields: list[str] | None = None,
220 max_rows: int = 10_000,
221 ) -> None:
222 self._data_source = data_source
223 self._required_fields: list[str] = required_fields or []
224 self._max_rows = max_rows
226 # ------------------------------------------------------------------
227 # Public API
228 # ------------------------------------------------------------------
230 async def parse(
231 self,
232 content: bytes,
233 filename: str,
234 *,
235 column_map: dict[str, str] | None = None,
236 ) -> Result[ImportJob, AdminError]:
237 """Parse uploaded file bytes and return a validated ImportJob.
239 Validation is non-destructive — nothing is written to the data source.
241 Args:
242 content: Raw file bytes.
243 filename: Original filename; used to detect format (csv / json).
244 column_map: Optional explicit header-to-field mapping.
245 CSV: ``{"CSV Header": "model_field"}``
246 JSON: ``{"json_key": "model_field"}``
247 Defaults to identity mapping (header == field).
249 Returns:
250 ``Result[ImportJob, AdminError]`` — Ok even when rows have
251 validation errors (so callers can show the preview).
252 Err only on file-level failures (wrong format, size limit).
253 """
254 lower = filename.lower()
255 if lower.endswith(".csv"):
256 rows, effective_map, parse_errors = _parse_csv(
257 content, column_map=column_map
258 )
259 elif lower.endswith((".json", ".jsonl")):
260 rows, effective_map, parse_errors = _parse_json(
261 content, column_map=column_map
262 )
263 else:
264 return Err(
265 AdminError(
266 message=f"Unsupported file format: {filename!r}. Use .csv or .json."
267 )
268 )
270 if parse_errors and not rows:
271 return Err(AdminError(message=parse_errors[0].message))
273 if self._max_rows and len(rows) > self._max_rows:
274 return Err(
275 AdminError(
276 message=f"File contains {len(rows):,} rows which exceeds the limit of {self._max_rows:,}."
277 )
278 )
280 # Row-level validation
281 validation_errors = list(parse_errors)
282 validation_errors.extend(self._validate_rows(rows))
284 job = ImportJob(
285 rows=rows,
286 errors=validation_errors,
287 column_map=effective_map,
288 source_filename=filename,
289 total_rows=len(rows),
290 )
291 logger.info(
292 "Import parsed: filename=%s rows=%d errors=%d",
293 filename,
294 len(rows),
295 len(validation_errors),
296 )
297 return Ok(job)
299 async def commit(self, job: ImportJob) -> Result[ImportResult, AdminError]:
300 """Write valid rows from *job* to the data source.
302 Rows with errors are skipped. Remaining rows are inserted one at a
303 time; individual insert failures are collected and do not abort the
304 rest of the batch.
306 Args:
307 job: A parsed ImportJob (from :meth:`parse`).
309 Returns:
310 ``Result[ImportResult, AdminError]`` — Ok with counts and any
311 per-row commit errors.
312 """
313 created = 0
314 failed = 0
315 commit_errors: list[ImportRowError] = []
316 error_row_set = {e.row for e in job.errors}
318 for row_num, row in enumerate(job.rows, start=1):
319 if row_num in error_row_set:
320 failed += 1
321 continue
322 try:
323 await self._data_source.create(row)
324 created += 1
325 except (ValueError, TypeError, KeyError, RuntimeError) as exc:
326 failed += 1
327 commit_errors.append(
328 ImportRowError(row=row_num, field="__row__", message=str(exc))
329 )
330 logger.warning("Import commit error at row %d: %s", row_num, exc)
332 result = ImportResult(created=created, failed=failed, errors=commit_errors)
333 logger.info(
334 "Import committed: filename=%s created=%d failed=%d",
335 job.source_filename,
336 created,
337 failed,
338 )
339 return Ok(result)
341 # ------------------------------------------------------------------
342 # Internal helpers
343 # ------------------------------------------------------------------
345 def _validate_rows(self, rows: list[dict[str, Any]]) -> list[ImportRowError]:
346 """Run required-field validation over all rows.
348 Args:
349 rows: Parsed rows to validate.
351 Returns:
352 List of ImportRowError for any violations found.
353 """
354 errors: list[ImportRowError] = []
355 for row_num, row in enumerate(rows, start=1):
356 for field_name in self._required_fields:
357 val = row.get(field_name)
358 if val is None or (isinstance(val, str) and not val.strip()):
359 errors.append(
360 ImportRowError(
361 row=row_num,
362 field=field_name,
363 message=f"'{field_name}' is required",
364 )
365 )
366 return errors
369__all__ = [
370 "AdminImportService",
371 "ImportJob",
372 "ImportResult",
373 "ImportRowError",
374]