"""Export/Import der Stammdaten als JSON – mit ID-Erhalt für Sicherung/Umzug. Umfasst Kategorien (inkl. eigener Felder), Lagerorte, Einheiten und Gebinde. Die IDs werden bewusst mitgeführt: nur so passen z.B. gedruckte Lagerort-QR-Codes (`/l/`) nach einem Wiederherstellen noch, und Verweise (parent_id, category_id) bleiben gültig. Beim Import entscheidet der Modus, was bei bereits vergebener ID passiert: - ``skip`` – vorhandene Einträge unangetastet lassen, nur neue anlegen. - ``overwrite`` – vorhandene per ID überschreiben, fehlende anlegen. Einzelne Zeilen, die an einer Eindeutigkeit scheitern (z.B. gleicher Name unter anderer ID), werden per Savepoint übersprungen statt den ganzen Import abzubrechen. """ from __future__ import annotations import json from sqlalchemy import text from sqlalchemy.orm import Session from ..models import Category, FieldDefinition, Location, PackageType, Unit, UnitKind from .fields import options_list FORMAT_KEY = "vorrania_master_data" FORMAT_VERSION = 1 _TABLES = ("units", "package_types", "categories", "field_definitions", "locations") def export_master_data(db: Session) -> dict: return { FORMAT_KEY: FORMAT_VERSION, "categories": [ {"id": c.id, "name": c.name, "parent_id": c.parent_id, "tracking": c.tracking} for c in db.query(Category).order_by(Category.id).all() ], "field_definitions": [ { "id": f.id, "category_id": f.category_id, "label": f.label, "key": f.key, "field_type": f.field_type, "unit": f.unit, "options": options_list(f), "required": f.required, "position": f.position, "is_builtin": f.is_builtin, } for f in db.query(FieldDefinition).order_by(FieldDefinition.id).all() ], "locations": [ {"id": l.id, "name": l.name, "parent_id": l.parent_id} for l in db.query(Location).order_by(Location.id).all() ], "units": [ {"id": u.id, "name": u.name, "kind": u.kind.value, "factor": u.factor, "is_builtin": u.is_builtin} for u in db.query(Unit).order_by(Unit.id).all() ], "package_types": [ {"id": p.id, "singular": p.singular, "plural": p.plural, "is_builtin": p.is_builtin} for p in db.query(PackageType).order_by(PackageType.id).all() ], } class _Report: def __init__(self) -> None: self.created: dict[str, int] = {} self.updated: dict[str, int] = {} self.skipped: dict[str, int] = {} def add(self, bucket: dict[str, int], key: str) -> None: bucket[key] = bucket.get(key, 0) + 1 def as_dict(self) -> dict: return {"created": self.created, "updated": self.updated, "skipped": self.skipped} def _fix_sequence(db: Session, table: str) -> None: """Postgres: die id-Sequenz auf MAX(id) heben, sonst kollidieren spätere Auto-IDs mit importierten. SQLite braucht das nicht.""" if db.get_bind().dialect.name != "postgresql": return db.execute(text( f"SELECT setval(pg_get_serial_sequence('{table}', 'id'), " f"GREATEST((SELECT COALESCE(MAX(id), 1) FROM {table}), 1))" )) def import_master_data(db: Session, data: dict, mode: str = "skip") -> dict: if not isinstance(data, dict) or FORMAT_KEY not in data: raise ValueError("Das ist keine Vorrania-Stammdaten-Datei.") overwrite = mode == "overwrite" rep = _Report() def upsert(model, row, key, build, *, is_tree=False, conflict=None) -> None: rid = row.get("id") existing = db.get(model, rid) if rid is not None else None if existing is not None and not overwrite: rep.add(rep.skipped, key) return # Vorab-Check statt IntegrityError abzufangen: würde die Zeile eine # Eindeutigkeit unter ANDERER id verletzen (z.B. gleicher Name), lieber # überspringen als den Import abbrechen. if conflict is not None and conflict(row): rep.add(rep.skipped, key) return if existing is not None: for k, v in build(row).items(): setattr(existing, k, v) db.flush() rep.add(rep.updated, key) return kwargs = build(row) if rid is not None: kwargs["id"] = rid if is_tree: kwargs["parent_id"] = None # Eltern erst im 2. Durchgang db.add(model(**kwargs)) db.flush() rep.add(rep.created, key) def set_parents(model, rows) -> None: for row in rows or []: rid = row.get("id") obj = db.get(model, rid) if rid is not None else None if obj is None: continue pid = row.get("parent_id") ziel = db.get(model, pid) if pid is not None else None obj.parent_id = ziel.id if ziel is not None else None db.flush() def name_clash(model, field, r): val = r.get(field) return db.query(model.id).filter( getattr(model, field) == val, model.id != r.get("id") ).first() is not None for row in data.get("units") or []: upsert(Unit, row, "units", lambda r: { "name": r["name"], "kind": UnitKind(r["kind"]), "factor": float(r.get("factor", 1.0)), "is_builtin": bool(r.get("is_builtin", False)), }, conflict=lambda r: name_clash(Unit, "name", r)) for row in data.get("package_types") or []: upsert(PackageType, row, "package_types", lambda r: { "singular": r["singular"], "plural": r.get("plural") or r["singular"], "is_builtin": bool(r.get("is_builtin", False)), }, conflict=lambda r: name_clash(PackageType, "singular", r)) cats = data.get("categories") or [] for row in cats: upsert(Category, row, "categories", lambda r: { "name": r["name"], "tracking": r.get("tracking") or "food", }, is_tree=True) set_parents(Category, cats) for row in data.get("field_definitions") or []: if db.get(Category, row.get("category_id")) is None: rep.add(rep.skipped, "field_definitions") continue upsert(FieldDefinition, row, "field_definitions", lambda r: { "category_id": r["category_id"], "label": r["label"], "key": r.get("key") or r["label"], "field_type": r.get("field_type", "text"), "unit": r.get("unit"), "options": json.dumps(r.get("options") or []) if r.get("field_type") == "select" else None, "required": bool(r.get("required", False)), "position": int(r.get("position", 0)), "is_builtin": bool(r.get("is_builtin", False)), }) locs = data.get("locations") or [] for row in locs: upsert(Location, row, "locations", lambda r: {"name": r["name"]}, is_tree=True) set_parents(Location, locs) db.commit() for table in _TABLES: _fix_sequence(db, table) db.commit() return rep.as_dict()