116 lines
No EOL
4.1 KiB
Python
116 lines
No EOL
4.1 KiB
Python
from typing import List, Optional, TYPE_CHECKING
|
||
if TYPE_CHECKING:
|
||
from .inventory import Inventory
|
||
|
||
from sqlalchemy import Identity, Integer, Unicode
|
||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||
|
||
from . import db
|
||
from ..temp import is_temp_id
|
||
from ..utils.validation import ValidatableMixin
|
||
|
||
class Item(ValidatableMixin, db.Model):
|
||
__tablename__ = 'Items'
|
||
VALIDATION_LABEL = 'Item'
|
||
|
||
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
|
||
description: Mapped[Optional[str]] = mapped_column("Description", Unicode(255), nullable=True)
|
||
category: Mapped[Optional[str]] = mapped_column("Category", Unicode(255), nullable=True)
|
||
|
||
inventory: Mapped[List['Inventory']] = relationship('Inventory', back_populates='item')
|
||
|
||
def __init__(self, description: Optional[str] = None, category: Optional[str] = None):
|
||
self.description = description
|
||
self.category = category
|
||
|
||
def __repr__(self):
|
||
return f"<Item(id={self.id}, description={repr(self.description)}, category={repr(self.category)})>"
|
||
|
||
def serialize(self):
|
||
return {
|
||
'id': self.id,
|
||
'name': self.description,
|
||
'category': self.category
|
||
}
|
||
|
||
@classmethod
|
||
def sync_from_state(cls, submitted_items: list[dict]) -> dict[str, int]:
|
||
submitted_clean = []
|
||
seen_ids = set()
|
||
temp_id_map = {}
|
||
|
||
for item in submitted_items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
name = str(item.get("name", "")).strip()
|
||
raw_id = item.get("id")
|
||
|
||
if not name:
|
||
continue
|
||
|
||
try:
|
||
if raw_id:
|
||
parsed_id = int(raw_id)
|
||
if parsed_id >= 0:
|
||
seen_ids.add(parsed_id)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
submitted_clean.append({"id": raw_id, "description": name})
|
||
|
||
existing_by_id = {t.id: t for t in db.session.query(cls).all()}
|
||
existing_ids = set(existing_by_id.keys())
|
||
|
||
print(f"Existing item IDs: {existing_ids}")
|
||
print(f"Submitted item IDs: {seen_ids}")
|
||
|
||
for entry in submitted_clean:
|
||
submitted_id = entry["id"]
|
||
description = entry["description"]
|
||
|
||
if is_temp_id(submitted_id):
|
||
obj = cls(description=description)
|
||
db.session.add(obj)
|
||
db.session.flush()
|
||
temp_id_map[submitted_id] = obj.id
|
||
print(f"➕ Adding type: {description}")
|
||
elif isinstance(submitted_id, int) or submitted_id.isdigit():
|
||
submitted_id_int = int(submitted_id)
|
||
obj = existing_by_id.get(submitted_id_int)
|
||
if obj and obj.description != description:
|
||
print(f"✏️ Updating type {obj.id}: '{obj.description}' → '{description}'")
|
||
obj.description = description
|
||
|
||
for id_to_remove in existing_ids - seen_ids:
|
||
obj = existing_by_id[id_to_remove]
|
||
db.session.delete(obj)
|
||
print(f"🗑️ Removing type ID {id_to_remove}")
|
||
|
||
id_map = {
|
||
**{str(i): i for i in seen_ids},
|
||
**{str(temp): real for temp, real in temp_id_map.items()}
|
||
}
|
||
return id_map
|
||
|
||
@classmethod
|
||
def validate_state(cls, submitted_items: list[dict]) -> list[str]:
|
||
errors = []
|
||
|
||
for index, item in enumerate(submitted_items):
|
||
if not isinstance(item, dict):
|
||
errors.append(f"Area entry #{index + 1} is not a valid object.")
|
||
continue
|
||
|
||
name = item.get('name')
|
||
if not name or not str(name).strip():
|
||
errors.append(f"Area entry #{index + 1} is missing a name.")
|
||
|
||
raw_id = item.get('id')
|
||
if raw_id is not None:
|
||
try:
|
||
_ = int(raw_id)
|
||
except (ValueError, TypeError):
|
||
if not is_temp_id(raw_id):
|
||
errors.append(f"Area entry #{index + 1} has invalid ID: {raw_id}")
|
||
|
||
return errors |