117 lines
4 KiB
Python
117 lines
4 KiB
Python
from typing import List, Optional, TYPE_CHECKING
|
|
if TYPE_CHECKING:
|
|
from .rooms import Room
|
|
|
|
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 Area(ValidatableMixin, db.Model):
|
|
__tablename__ = 'area'
|
|
VALIDATION_LABEL = "Area"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True)
|
|
name: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True)
|
|
|
|
rooms: Mapped[List['Room']] = relationship('Room', back_populates='area')
|
|
|
|
def __init__(self, name: Optional[str] = None):
|
|
self.name = name
|
|
|
|
def __repr__(self):
|
|
return f"<Area(id={self.id}, name={repr(self.name)})>"
|
|
|
|
def serialize(self):
|
|
return {
|
|
'id': self.id,
|
|
'name': self.name
|
|
}
|
|
|
|
@classmethod
|
|
def sync_from_state(cls, submitted_items: list[dict]) -> dict[str, int]:
|
|
"""
|
|
Syncs the Area table (aka 'sections') with the submitted list.
|
|
Supports add, update, and delete.
|
|
Also returns a mapping of temp IDs to real IDs for resolution.
|
|
"""
|
|
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
|
|
submitted_clean.append({"id": raw_id, "name": name})
|
|
|
|
# Record real (non-temp) IDs
|
|
try:
|
|
if raw_id is not None:
|
|
parsed_id = int(raw_id)
|
|
if parsed_id >= 0:
|
|
seen_ids.add(parsed_id)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
existing_query = db.session.query(cls)
|
|
existing_by_id = {area.id: area for area in existing_query.all()}
|
|
existing_ids = set(existing_by_id.keys())
|
|
|
|
for entry in submitted_clean:
|
|
submitted_id_raw = entry.get("id")
|
|
submitted_name = entry["name"]
|
|
|
|
if is_temp_id(submitted_id_raw):
|
|
new_area = cls(name=submitted_name)
|
|
db.session.add(new_area)
|
|
db.session.flush() # Get the real ID
|
|
temp_id_map[submitted_id_raw] = new_area.id
|
|
else:
|
|
try:
|
|
submitted_id = int(submitted_id_raw)
|
|
except (ValueError, TypeError):
|
|
continue # Skip malformed ID
|
|
|
|
if submitted_id in existing_by_id:
|
|
area = existing_by_id[submitted_id]
|
|
if area.name != submitted_name:
|
|
area.name = submitted_name
|
|
|
|
for existing_id in existing_ids - seen_ids:
|
|
area = existing_by_id[existing_id]
|
|
db.session.delete(area)
|
|
|
|
id_map = {
|
|
**{str(i): i for i in seen_ids}, # "1" → 1
|
|
**{str(temp): real for temp, real in temp_id_map.items()} # "temp-1" → 5
|
|
}
|
|
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
|