"""Orchestrates upload → analyze → map → validate → commit without silent master-data create.""" from __future__ import annotations from collections import defaultdict from datetime import datetime, timezone from typing import Any from uuid import uuid4 import pandas as pd from sqlalchemy import select from sqlalchemy.orm import Session from db.models_v2 import V2ImportMapping, V2StagingBatch, V2StagingRow from db.v2_import.catalog import SqlAlchemyCatalogLookup, known_target_for from db.v2_import.commit import CommitError, commit_import from db.v2_import.constants import ( AUTO_RESOLVED_STATUSES, BLOCKING_STATUSES, ENTITY_TYPE_LABELS, IGNORE_ALLOWED_ENTITY_TYPES, OPTIONAL_ENTITY_TYPES, RESOLVED_STATUSES, SELECTABLE_ATTRIBUTE_TYPES, UI_STATUS_LABELS, ) from db.v2_import.matching import ( CatalogEntity, classify_match, display_lookup_value, normalize_lookup_value, similar_entities, ) from db.v2_import.parser import parse_product_csv_dataframe class ImportMapperError(ValueError): pass def _utcnow() -> datetime: return datetime.now(timezone.utc) def _mapping_key(entity_type: str, source_value: str, parent_key: str = "") -> tuple[str, str, str]: return (entity_type, normalize_lookup_value(source_value), parent_key or "") class ProductImportService: def __init__(self, session: Session, *, catalog: SqlAlchemyCatalogLookup | None = None) -> None: self.session = session self.catalog = catalog or SqlAlchemyCatalogLookup(session) def upload_csv( self, df: pd.DataFrame, *, filename: str, uploaded_by: str, source_profile: str = "", ) -> V2StagingBatch: parsed = parse_product_csv_dataframe(df) batch = V2StagingBatch( id=str(uuid4()), filename=filename or "products.csv", uploaded_by=uploaded_by or "dashboard", row_count=len(parsed.rows), status="uploaded", source_profile=source_profile or "", summary_json={}, created_at=_utcnow(), updated_at=_utcnow(), ) self.session.add(batch) self.session.flush() if parsed.missing_required: batch.status = "failed" batch.error_message = f"Missing required column: {', '.join(parsed.missing_required)}" batch.updated_at = _utcnow() self.session.flush() return batch for row in parsed.rows: self.session.add( V2StagingRow( batch_id=batch.id, row_number=row.row_number, raw_data=row.raw, normalized_data=row.as_normalized(), status="skipped" if row.error else "pending", blocking_fields=["sku"] if row.error else None, created_at=_utcnow(), updated_at=_utcnow(), ) ) self.session.flush() self.analyze(batch) return batch def analyze(self, batch: V2StagingBatch) -> V2StagingBatch: batch.status = "analyzing" batch.error_message = None batch.updated_at = _utcnow() self.session.flush() rows = self._rows(batch.id) parsed_rows = [self._row_view(row) for row in rows] existing_by_key = { (item.entity_type, item.normalized_value, item.parent_key or ""): item for item in self._mappings(batch.id) } self._analyze_lookup_role(batch, parsed_rows, existing_by_key, "brand") self._analyze_lookup_role(batch, parsed_rows, existing_by_key, "hub") self._analyze_lookup_role(batch, parsed_rows, existing_by_key, "category") self._analyze_lookup_role(batch, parsed_rows, existing_by_key, "collection") self._analyze_attributes(batch, parsed_rows, existing_by_key) self._analyze_attribute_options(batch, parsed_rows, existing_by_key) self._refresh_row_states(batch) self._refresh_session_state(batch) return batch def resolve_mapping( self, mapping: V2ImportMapping, *, action: str, resolved_by: str, target_id: int | None = None, create_payload: dict | None = None, confirm_similar: bool = False, persist_mapping: bool = True, ) -> V2ImportMapping: batch = self._batch(mapping.batch_id) if batch.status in {"importing", "completed", "cancelled"}: raise ImportMapperError("This import can no longer be mapped.") action = str(action or "").strip().lower() now = _utcnow() if action == "ignore": if mapping.entity_type not in IGNORE_ALLOWED_ENTITY_TYPES: raise ImportMapperError(f"{ENTITY_TYPE_LABELS[mapping.entity_type]} cannot be ignored.") mapping.status = "ignored" mapping.matched_target_id = None mapping.matched_target_name = None mapping.create_payload = None mapping.resolution_source = "user_ignore" elif action == "map": if target_id is None: raise ImportMapperError("Select an existing value to map to.") target = self._load_target(mapping.entity_type, int(target_id), mapping=mapping) if target is None: raise ImportMapperError("The selected record was not found or is the wrong type.") mapping.status = "mapped" mapping.matched_target_id = target.id mapping.matched_target_name = target.name mapping.create_payload = None mapping.resolution_source = "user_map" if persist_mapping: self.catalog.remember_mapping( entity_type=mapping.entity_type, source_profile=batch.source_profile or "", source_value=mapping.source_value, normalized_value=mapping.normalized_value, parent_key=mapping.parent_key or "", target_id=target.id, target_name=target.name, created_by=resolved_by, ) elif action == "create": payload = dict(create_payload or {}) payload.setdefault("name", mapping.source_value) similar = self._similar_for_create(mapping, payload) if similar and not confirm_similar: mapping.create_payload = payload raise ImportMapperError( f"A similar {ENTITY_TYPE_LABELS[mapping.entity_type]} already exists. " "Map to the existing value unless you are sure this is new." ) created = self._create_entity(mapping, payload, created_by=resolved_by) mapping.status = "created" mapping.matched_target_id = created.id mapping.matched_target_name = created.name mapping.create_payload = payload mapping.resolution_source = "user_create" if persist_mapping: self.catalog.remember_mapping( entity_type=mapping.entity_type, source_profile=batch.source_profile or "", source_value=mapping.source_value, normalized_value=mapping.normalized_value, parent_key=mapping.parent_key or "", target_id=created.id, target_name=created.name, created_by=resolved_by, ) else: raise ImportMapperError("Choose Map to existing, Create new, or Ignore.") mapping.resolved_by = resolved_by mapping.resolved_at = now mapping.updated_at = now if mapping.entity_type == "attribute": self._refresh_options_for_attribute(batch, mapping) self.session.flush() self._refresh_row_states(batch) self._refresh_session_state(batch) return mapping def bulk_resolve( self, mapping_ids: list[int], *, action: str, resolved_by: str, target_id: int | None = None, persist_mapping: bool = True, ) -> int: count = 0 for mapping_id in mapping_ids: mapping = self.session.get(V2ImportMapping, int(mapping_id)) if mapping is None: continue self.resolve_mapping( mapping, action=action, resolved_by=resolved_by, target_id=target_id, persist_mapping=persist_mapping, ) count += 1 return count def session_payload(self, batch: V2StagingBatch) -> dict[str, Any]: mappings = self._mappings(batch.id) counts = defaultdict(int) unresolved_by_type = defaultdict(int) creates_by_type = defaultdict(int) for item in mappings: counts[item.status] += 1 if item.status in BLOCKING_STATUSES: unresolved_by_type[item.entity_type] += 1 if item.status == "created": creates_by_type[item.entity_type] += 1 rows = self._rows(batch.id) valid_rows = [row for row in rows if not (row.normalized_data or {}).get("error")] skus = [str((row.normalized_data or {}).get("sku") or "") for row in valid_rows] existing = self.catalog.existing_skus([sku for sku in skus if sku]) unresolved = sum(unresolved_by_type.values()) gate = "ready" if unresolved == 0 and batch.status not in {"failed", "cancelled"} else "requires_mapping" if any(item.status == "invalid" for item in mappings): gate = "invalid" if batch.status in {"failed", "cancelled"}: gate = batch.status summary = { "total_product_rows": len(rows), "valid_rows": len(valid_rows), "invalid_rows": len(rows) - len(valid_rows), "new_products": max(0, len({sku for sku in skus if sku}) - len(existing)), "existing_products": len(existing), "exact_mappings": counts["exact_match"], "saved_mappings": counts["known_mapping"], "suggested_mappings": counts["suggested"] + counts["ambiguous"], "unresolved_mappings": unresolved, "new_entities_to_create": dict(creates_by_type), "unresolved_by_type": dict(unresolved_by_type), "status_counts": dict(counts), "gate": gate, } batch.summary_json = summary batch.updated_at = _utcnow() return { "id": str(batch.id), "filename": batch.filename, "uploaded_by": batch.uploaded_by, "source_profile": batch.source_profile or "", "status": batch.status, "row_count": batch.row_count, "error_message": batch.error_message, "summary": summary, "result": batch.result_json, "created_at": batch.created_at.isoformat() if batch.created_at else None, "updated_at": batch.updated_at.isoformat() if batch.updated_at else None, } def mapping_payload(self, item: V2ImportMapping) -> dict[str, Any]: return { "id": int(item.id), "entity_type": item.entity_type, "entity_label": ENTITY_TYPE_LABELS.get(item.entity_type, item.entity_type), "source_value": item.source_value, "normalized_value": item.normalized_value, "occurrence_count": item.occurrence_count, "status": item.status, "status_label": UI_STATUS_LABELS.get(item.status, item.status), "matched_target_id": int(item.matched_target_id) if item.matched_target_id is not None else None, "matched_target_name": item.matched_target_name, "suggested_targets": item.suggested_targets or [], "parent_entity_type": item.parent_entity_type, "parent_source_value": item.parent_source_value, "parent_target_id": int(item.parent_target_id) if item.parent_target_id is not None else None, "is_optional": bool(item.is_optional), "can_ignore": item.entity_type in IGNORE_ALLOWED_ENTITY_TYPES, "create_payload": item.create_payload, "resolution_source": item.resolution_source, "resolved_by": item.resolved_by, "resolved_at": item.resolved_at.isoformat() if item.resolved_at else None, } def preview_rows(self, batch: V2StagingBatch, *, limit: int = 12) -> list[dict[str, Any]]: resolved_index = self._resolved_index(batch.id) out = [] for row in self._rows(batch.id)[:limit]: data = dict(row.normalized_data or {}) out.append( { "row_number": row.row_number, "sku": data.get("sku"), "title": data.get("title"), "incoming": { "brand": data.get("brand") or "", "hub": data.get("hub") or "", "category": data.get("category") or "", "collection": data.get("collection") or "", }, "resolved": self._resolved_lookups(data, resolved_index), "status": row.status, "blocking_fields": row.blocking_fields or [], } ) return out def commit(self, batch: V2StagingBatch, *, committed_by: str) -> dict[str, Any]: self._refresh_row_states(batch) self._refresh_session_state(batch) if batch.status != "ready": summary = batch.summary_json or {} unresolved = int(summary.get("unresolved_mappings") or 0) raise ImportMapperError( f"{unresolved} values must be reviewed before products can be imported." if unresolved else "This import is not ready to run." ) batch.status = "importing" batch.updated_at = _utcnow() self.session.flush() try: result = commit_import(self.session, batch, catalog=self.catalog, committed_by=committed_by) except CommitError as exc: batch.status = "failed" batch.error_message = str(exc) batch.updated_at = _utcnow() self.session.flush() raise ImportMapperError(str(exc)) from exc batch.status = "completed" batch.result_json = result batch.updated_at = _utcnow() self.session.flush() return result def cancel(self, batch: V2StagingBatch) -> V2StagingBatch: if batch.status in {"completed", "importing"}: raise ImportMapperError("A finished or running import cannot be cancelled.") batch.status = "cancelled" batch.updated_at = _utcnow() self.session.flush() return batch def search_entities( self, entity_type: str, query: str, *, hub_id: int | None = None, attribute_id: int | None = None, ) -> list[dict[str, Any]]: items = self.catalog.search( entity_type, query, hub_id=hub_id, attribute_id=attribute_id, ) return [ { "id": item.id, "name": item.name, "code": item.code, "parent_id": item.parent_id, "parent_name": item.parent_name, "label": item.context_label, "data_type": (item.extra or {}).get("data_type"), } for item in items ] def _analyze_lookup_role( self, batch: V2StagingBatch, parsed_rows: list[dict[str, Any]], existing_by_key: dict[tuple[str, str, str], V2ImportMapping], role: str, ) -> None: counts: dict[str, int] = defaultdict(int) for row in parsed_rows: if row.get("error"): continue value = display_lookup_value(row.get(role) or "") if value: counts[value] += 1 candidates = self._candidates_for(role) for source_value, occurrence_count in counts.items(): self._upsert_mapping( batch, existing_by_key, entity_type=role, source_value=source_value, occurrence_count=occurrence_count, candidates=candidates, is_optional=role in OPTIONAL_ENTITY_TYPES, ) def _analyze_attributes( self, batch: V2StagingBatch, parsed_rows: list[dict[str, Any]], existing_by_key: dict[tuple[str, str, str], V2ImportMapping], ) -> None: counts: dict[str, int] = defaultdict(int) for row in parsed_rows: if row.get("error"): continue for header in (row.get("attributes") or {}): counts[str(header)] += 1 candidates = self.catalog.list_attributes() for source_value, occurrence_count in counts.items(): self._upsert_mapping( batch, existing_by_key, entity_type="attribute", source_value=source_value, occurrence_count=occurrence_count, candidates=candidates, is_optional=True, ) def _analyze_attribute_options( self, batch: V2StagingBatch, parsed_rows: list[dict[str, Any]], existing_by_key: dict[tuple[str, str, str], V2ImportMapping], ) -> None: attribute_maps = { item.source_value: item for item in self._mappings(batch.id) if item.entity_type == "attribute" } values_by_header: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int)) for row in parsed_rows: if row.get("error"): continue for header, value in (row.get("attributes") or {}).items(): text = display_lookup_value(value) if text: values_by_header[str(header)][text] += 1 for header, value_counts in values_by_header.items(): attr_mapping = attribute_maps.get(header) if attr_mapping is None: continue attr_entity = None if attr_mapping.matched_target_id is not None: attr_entity = self.catalog.get_attribute(int(attr_mapping.matched_target_id)) data_type = (attr_entity.extra or {}).get("data_type") if attr_entity else None if attr_mapping.status in AUTO_RESOLVED_STATUSES | {"mapped", "created"}: if data_type not in SELECTABLE_ATTRIBUTE_TYPES: continue elif attr_mapping.status == "ignored": continue option_candidates = ( self.catalog.list_attribute_options(int(attr_entity.id)) if attr_entity else [] ) parent_key = normalize_lookup_value(header) for source_value, occurrence_count in value_counts.items(): mapping = self._upsert_mapping( batch, existing_by_key, entity_type="attribute_option", source_value=source_value, occurrence_count=occurrence_count, candidates=option_candidates, is_optional=True, parent_key=parent_key, parent_entity_type="attribute", parent_source_value=header, parent_target_id=int(attr_entity.id) if attr_entity else None, ) if attr_mapping.status in BLOCKING_STATUSES: mapping.status = "invalid" mapping.suggested_targets = [] mapping.resolution_source = "attribute_unresolved" mapping.updated_at = _utcnow() def _upsert_mapping( self, batch: V2StagingBatch, existing_by_key: dict[tuple[str, str, str], V2ImportMapping], *, entity_type: str, source_value: str, occurrence_count: int, candidates: list[CatalogEntity], is_optional: bool, parent_key: str = "", parent_entity_type: str | None = None, parent_source_value: str | None = None, parent_target_id: int | None = None, ) -> V2ImportMapping: key = _mapping_key(entity_type, source_value, parent_key) mapping = existing_by_key.get(key) known = None if mapping is None or mapping.status not in {"mapped", "created", "ignored"}: known = known_target_for( self.catalog, entity_type=entity_type, source_value=source_value, source_profile=batch.source_profile or "", parent_key=parent_key, ) if known is not None and entity_type in {"category", "collection"} and parent_target_id: scoped = [item for item in candidates if item.parent_id == parent_target_id] if scoped: candidates = scoped decision = classify_match(source_value, candidates, known_target=known) else: decision = None if mapping is None: mapping = V2ImportMapping( batch_id=batch.id, entity_type=entity_type, source_value=source_value, normalized_value=key[1], parent_key=parent_key or "", parent_entity_type=parent_entity_type, parent_source_value=parent_source_value, parent_target_id=parent_target_id, occurrence_count=occurrence_count, is_optional=is_optional, created_at=_utcnow(), updated_at=_utcnow(), ) self.session.add(mapping) existing_by_key[key] = mapping else: mapping.occurrence_count = occurrence_count mapping.parent_target_id = parent_target_id or mapping.parent_target_id mapping.updated_at = _utcnow() if decision is not None and mapping.status not in {"mapped", "created", "ignored"}: mapping.status = decision.status mapping.resolution_source = decision.resolution_source if decision.target is not None: mapping.matched_target_id = decision.target.id mapping.matched_target_name = decision.target.name else: mapping.matched_target_id = None mapping.matched_target_name = None mapping.suggested_targets = [ { "id": item.id, "name": item.name, "code": item.code, "parent_id": item.parent_id, "parent_name": item.parent_name, "label": item.context_label, } for item in decision.suggestions ] if decision.status in AUTO_RESOLVED_STATUSES: mapping.resolved_at = _utcnow() mapping.resolved_by = "system" self.session.flush() return mapping def _refresh_options_for_attribute(self, batch: V2StagingBatch, attr_mapping: V2ImportMapping) -> None: rows = [self._row_view(row) for row in self._rows(batch.id)] existing_by_key = { (item.entity_type, item.normalized_value, item.parent_key or ""): item for item in self._mappings(batch.id) } self._analyze_attribute_options(batch, rows, existing_by_key) def _refresh_row_states(self, batch: V2StagingBatch) -> None: resolved_index = self._resolved_index(batch.id) for row in self._rows(batch.id): data = dict(row.normalized_data or {}) if data.get("error"): row.status = "skipped" row.blocking_fields = ["sku"] continue blocking = [] for role in ("brand", "hub", "category", "collection"): value = display_lookup_value(data.get(role) or "") if not value: continue item = resolved_index.get(_mapping_key(role, value)) if item is None or item.status in BLOCKING_STATUSES: blocking.append(role) for header, value in (data.get("attributes") or {}).items(): attr_item = resolved_index.get(_mapping_key("attribute", header)) if attr_item is None or attr_item.status in BLOCKING_STATUSES: blocking.append(header) continue if attr_item.status == "ignored": continue option_item = resolved_index.get( _mapping_key("attribute_option", value, normalize_lookup_value(header)) ) if option_item is not None and option_item.status in BLOCKING_STATUSES: blocking.append(f"{header}:{value}") row.blocking_fields = blocking or None row.resolved_data = { "lookups": self._resolved_lookups(data, resolved_index), "attributes": self._resolved_attributes(data, resolved_index), } row.status = "blocked" if blocking else "resolved" row.updated_at = _utcnow() self.session.flush() def _refresh_session_state(self, batch: V2StagingBatch) -> None: payload = self.session_payload(batch) gate = payload["summary"]["gate"] if batch.status in {"completed", "cancelled", "failed", "importing"}: return if gate == "ready": batch.status = "ready" elif gate == "invalid": batch.status = "needs_mapping" else: batch.status = "needs_mapping" batch.updated_at = _utcnow() self.session.flush() def _resolved_index(self, batch_id: str) -> dict[tuple[str, str, str], V2ImportMapping]: return { (item.entity_type, item.normalized_value, item.parent_key or ""): item for item in self._mappings(batch_id) } def _resolved_lookups(self, data: dict[str, Any], index: dict) -> dict[str, Any]: out = {} for role in ("brand", "hub", "category", "collection"): value = display_lookup_value(data.get(role) or "") item = index.get(_mapping_key(role, value)) if value else None out[role] = { "incoming": value, "status": item.status if item else ("blank" if not value else "unmatched"), "target_id": int(item.matched_target_id) if item and item.matched_target_id is not None else None, "target_name": item.matched_target_name if item else None, } return out def _resolved_attributes(self, data: dict[str, Any], index: dict) -> dict[str, Any]: out = {} for header, value in (data.get("attributes") or {}).items(): attr_item = index.get(_mapping_key("attribute", header)) option_item = index.get(_mapping_key("attribute_option", value, normalize_lookup_value(header))) out[header] = { "incoming": value, "attribute_id": int(attr_item.matched_target_id) if attr_item and attr_item.matched_target_id else None, "attribute_name": attr_item.matched_target_name if attr_item else None, "attribute_status": attr_item.status if attr_item else "unmatched", "option_id": int(option_item.matched_target_id) if option_item and option_item.matched_target_id else None, "option_name": option_item.matched_target_name if option_item else None, } return out def _candidates_for(self, role: str) -> list[CatalogEntity]: if role == "brand": return self.catalog.list_brands() if role in {"hub", "category", "collection"}: return self.catalog.list_nodes(role) if role == "attribute": return self.catalog.list_attributes() return [] def _load_target(self, entity_type: str, target_id: int, *, mapping: V2ImportMapping) -> CatalogEntity | None: if entity_type == "brand": return self.catalog.get_brand(target_id) if entity_type in {"hub", "category", "collection"}: return self.catalog.get_node(target_id, node_type=entity_type) if entity_type == "attribute": return self.catalog.get_attribute(target_id) if entity_type == "attribute_option": option = self.catalog.get_attribute_option(target_id) if option is None: return None if mapping.parent_target_id and option.parent_id != mapping.parent_target_id: return None return option return None def _similar_for_create(self, mapping: V2ImportMapping, payload: dict[str, Any]) -> list[dict[str, Any]]: name = str(payload.get("name") or mapping.source_value) if mapping.entity_type == "brand": return similar_entities(name, self.catalog.list_brands()) if mapping.entity_type in {"hub", "category", "collection"}: parent_id = payload.get("hub_id") or payload.get("parent_id") candidates = self.catalog.list_nodes( mapping.entity_type, parent_id=int(parent_id) if parent_id is not None else None, ) if mapping.entity_type != "collection" and parent_id is None: candidates = self.catalog.list_nodes(mapping.entity_type) return similar_entities(name, candidates) if mapping.entity_type == "attribute": return similar_entities(name, self.catalog.list_attributes()) if mapping.entity_type == "attribute_option" and mapping.parent_target_id: return similar_entities(name, self.catalog.list_attribute_options(int(mapping.parent_target_id))) return [] def _create_entity(self, mapping: V2ImportMapping, payload: dict[str, Any], *, created_by: str) -> CatalogEntity: from db.master_taxonomy_registry import registry_code_from_name, upsert_brand from db.models_v2 import V2Attribute, V2AttributeValue from db.v2_taxonomy import evaluate_taxonomy_creation, require_valid_taxonomy_parent name = display_lookup_value(payload.get("name") or mapping.source_value) if mapping.entity_type == "brand": if not name: raise ImportMapperError("Brand name is required.") created = upsert_brand( self.session, { "name": name, "code": payload.get("code") or registry_code_from_name(name), "is_active": True, "aliases": {"items": [mapping.source_value]}, }, ) entity = self.catalog.get_brand(int(created["id"])) if entity is None: raise ImportMapperError("The new Brand could not be loaded.") return entity if mapping.entity_type in {"hub", "category", "collection"}: parent_id = payload.get("parent_id") or payload.get("hub_id") parent_id = int(parent_id) if parent_id not in (None, "") else None try: require_valid_taxonomy_parent(self.session, mapping.entity_type, parent_id) except ValueError as exc: raise ImportMapperError(str(exc)) from exc outcome = evaluate_taxonomy_creation( self.session, name=name, node_type=mapping.entity_type, parent_id=parent_id, level=payload.get("level"), confirm_new=True, created_by=created_by, ) if outcome.status != "created" or outcome.category is None: raise ImportMapperError("That value could not be created. Map to an existing record instead.") entity = self.catalog.get_node(int(outcome.category.id), node_type=mapping.entity_type) if entity is None: raise ImportMapperError("The new record could not be loaded.") return entity if mapping.entity_type == "attribute": from db.v2_import.matching import attribute_code_from_header data_type = str(payload.get("data_type") or "").strip().lower() if not data_type: raise ImportMapperError("Choose an attribute type before creating it.") code = str(payload.get("code") or attribute_code_from_header(name)).strip().lower() row = V2Attribute( code=code, label=name, data_type=data_type, is_variant_axis=bool(payload.get("is_variant_axis", False)), is_channel_required=bool(payload.get("is_channel_required", True)), applies_to_product_types=list(payload.get("applies_to_product_types") or []), ) self.session.add(row) self.session.flush() entity = self.catalog.get_attribute(int(row.id)) if entity is None: raise ImportMapperError("The new Attribute could not be loaded.") return entity if mapping.entity_type == "attribute_option": attribute_id = mapping.parent_target_id or payload.get("attribute_id") if attribute_id is None: raise ImportMapperError("Map the Attribute first, then create this value.") from db.v2_import.matching import attribute_code_from_header row = V2AttributeValue( attribute_id=int(attribute_id), value_code=str(payload.get("code") or attribute_code_from_header(name)).strip().lower(), display_label=name, ) self.session.add(row) self.session.flush() entity = self.catalog.get_attribute_option(int(row.id)) if entity is None: raise ImportMapperError("The new attribute value could not be loaded.") return entity raise ImportMapperError("This value type cannot be created here.") def _batch(self, batch_id: str) -> V2StagingBatch: batch = self.session.get(V2StagingBatch, str(batch_id)) if batch is None: raise ImportMapperError("Import session not found.") return batch def _rows(self, batch_id: str) -> list[V2StagingRow]: return list( self.session.execute( select(V2StagingRow).where(V2StagingRow.batch_id == batch_id).order_by(V2StagingRow.row_number.asc()) ).scalars().all() ) def _mappings(self, batch_id: str) -> list[V2ImportMapping]: return list( self.session.execute( select(V2ImportMapping) .where(V2ImportMapping.batch_id == batch_id) .order_by(V2ImportMapping.entity_type.asc(), V2ImportMapping.source_value.asc()) ).scalars().all() ) def _row_view(self, row: V2StagingRow) -> dict[str, Any]: data = dict(row.normalized_data or {}) data["row_number"] = row.row_number return data