Lots and lots of unneeded code removed now!

This commit is contained in:
Yaro Kasear 2025-08-14 12:03:52 -05:00
parent 7854e9c910
commit 9705606c89
11 changed files with 11 additions and 620 deletions

View file

@ -6,12 +6,9 @@ from sqlalchemy import Identity, Integer, Unicode
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from . import db from . import db
from ..temp import is_temp_id
from ..utils.validation import ValidatableMixin
class Area(ValidatableMixin, db.Model): class Area(db.Model):
__tablename__ = 'area' __tablename__ = 'area'
VALIDATION_LABEL = "Area"
id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True) id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True)
name: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True) name: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True)
@ -23,95 +20,3 @@ class Area(ValidatableMixin, db.Model):
def __repr__(self): def __repr__(self):
return f"<Area(id={self.id}, name={repr(self.name)})>" 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

View file

@ -1,4 +1,4 @@
from typing import List, Optional, TYPE_CHECKING from typing import List, TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
from .inventory import Inventory from .inventory import Inventory
@ -6,12 +6,9 @@ from sqlalchemy import Identity, Integer, Unicode
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from . import db from . import db
from ..temp import is_temp_id
from ..utils.validation import ValidatableMixin
class Brand(ValidatableMixin, db.Model): class Brand(db.Model):
__tablename__ = 'brand' __tablename__ = 'brand'
VALIDATION_LABEL = 'Brand'
id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True) id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True)
name: Mapped[str] = mapped_column(Unicode(255), nullable=False) name: Mapped[str] = mapped_column(Unicode(255), nullable=False)
@ -24,90 +21,6 @@ class Brand(ValidatableMixin, db.Model):
def __repr__(self): def __repr__(self):
return f"<Brand(id={self.id}, name={repr(self.name)})>" return f"<Brand(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]:
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})
try:
if raw_id:
parsed_id = int(raw_id)
if parsed_id >= 0:
seen_ids.add(parsed_id)
except (ValueError, TypeError):
pass
existing_by_id = {b.id: b for b in db.session.query(cls).all()}
existing_ids = set(existing_by_id.keys())
for entry in submitted_clean:
submitted_id = entry.get("id")
name = entry["name"]
if is_temp_id(submitted_id):
obj = cls(name=name)
db.session.add(obj)
db.session.flush()
temp_id_map[submitted_id] = obj.id
else:
try:
parsed_id = int(submitted_id)
except (ValueError, TypeError):
continue
if parsed_id in existing_by_id:
obj = existing_by_id[parsed_id]
if obj.name != name:
obj.name = name
for id_to_remove in existing_ids - seen_ids:
db.session.delete(existing_by_id[id_to_remove])
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
@property @property
def identifier(self) -> str: def identifier(self) -> str:
return self.name if self.name else f"ID: {self.id}" return self.name if self.name else f"ID: {self.id}"

View file

@ -6,12 +6,9 @@ from sqlalchemy import Identity, Integer, Unicode
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from . import db from . import db
from ..temp import is_temp_id
from ..utils.validation import ValidatableMixin
class Item(ValidatableMixin, db.Model): class Item(db.Model):
__tablename__ = 'item' __tablename__ = 'item'
VALIDATION_LABEL = 'Item'
id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True) id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True)
description: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True) description: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True)
@ -24,88 +21,6 @@ class Item(ValidatableMixin, db.Model):
def __repr__(self): def __repr__(self):
return f"<Item(id={self.id}, description={repr(self.description)})>" return f"<Item(id={self.id}, description={repr(self.description)})>"
def serialize(self):
return {
'id': self.id,
'name': self.description
}
@property @property
def identifier(self): def identifier(self):
return self.description if self.description else f"Item {self.id}" return self.description if self.description else f"Item {self.id}"
@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())
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
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:
obj.description = description
for id_to_remove in existing_ids - seen_ids:
obj = existing_by_id[id_to_remove]
db.session.delete(obj)
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

View file

@ -6,12 +6,9 @@ from sqlalchemy import Identity, Integer, Unicode
from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.orm import Mapped, mapped_column, relationship
from . import db from . import db
from ..temp import is_temp_id
from ..utils.validation import ValidatableMixin
class RoomFunction(ValidatableMixin, db.Model): class RoomFunction(db.Model):
__tablename__ = 'room_function' __tablename__ = 'room_function'
VALIDATION_LABEL = "Function"
id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True) id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True)
description: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True) description: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True)
@ -23,62 +20,3 @@ class RoomFunction(ValidatableMixin, db.Model):
def __repr__(self): def __repr__(self):
return f"<RoomFunction(id={self.id}, description={repr(self.description)})>" return f"<RoomFunction(id={self.id}, description={repr(self.description)})>"
def serialize(self):
return {
'id': self.id,
'name': self.description
}
@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 = {f.id: f for f in db.session.query(cls).all()}
existing_ids = set(existing_by_id.keys())
for entry in submitted_clean:
submitted_id = entry.get("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
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:
obj.description = description
for id_to_remove in existing_ids - seen_ids:
obj = existing_by_id[id_to_remove]
db.session.delete(obj)
id_map = {
**{str(i): i for i in seen_ids},
**{str(temp): real for temp, real in temp_id_map.items()}
}
return id_map

View file

@ -10,11 +10,8 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship, joinedload, sele
from . import db from . import db
from ..utils.validation import ValidatableMixin class Room(db.Model):
class Room(ValidatableMixin, db.Model):
__tablename__ = 'rooms' __tablename__ = 'rooms'
VALIDATION_LABEL = "Room"
id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True) id: Mapped[int] = mapped_column(Integer, Identity(start=1, increment=1), primary_key=True)
name: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True) name: Mapped[Optional[str]] = mapped_column(Unicode(255), nullable=True)
@ -56,148 +53,6 @@ class Room(ValidatableMixin, db.Model):
func = self.room_function.description if self.room_function else "" func = self.room_function.description if self.room_function else ""
return f"{name} - {func}".strip(" -") return f"{name} - {func}".strip(" -")
def serialize(self):
return {
'id': self.id,
'name': self.name,
'area_id': self.area_id,
'function_id': self.function_id
}
@classmethod
def sync_from_state(cls, submitted_rooms: list[dict], section_map: dict[str, int], function_map: dict[str, int]) -> None:
"""
Syncs the Rooms table with the submitted room list.
Resolves foreign keys using section_map and function_map.
Supports add, update, and delete.
"""
def resolve_fk(key, fk_map, label):
if key is None:
return None
key = str(key)
if key.startswith("temp") or not key.isdigit():
if key in fk_map:
return fk_map[key]
raise ValueError(f"Unable to resolve {label} ID: {key}")
return int(key) # It's already a real ID
submitted_clean = []
seen_ids = set()
for room in submitted_rooms:
if not isinstance(room, dict):
continue
name = str(room.get("name", "")).strip()
if not name:
continue
rid = room.get("id")
area_id = room.get("area_id")
function_id = room.get("function_id")
submitted_clean.append({
"id": rid,
"name": name,
"area_id": area_id,
"function_id": function_id
})
if rid and not str(rid).startswith("room-"):
try:
seen_ids.add(int(rid))
except ValueError:
pass # Not valid? Not seen.
existing_query = db.session.query(cls)
existing_by_id = {room.id: room for room in existing_query.all()}
existing_ids = set(existing_by_id.keys())
for entry in submitted_clean:
rid = entry.get("id")
name = entry["name"]
resolved_area_id = resolve_fk(entry.get("area_id"), section_map, "section")
resolved_function_id = resolve_fk(entry.get("function_id"), function_map, "function")
if not rid or str(rid).startswith("room-"):
new_room = cls(name=name, area_id=resolved_area_id, function_id=resolved_function_id)
db.session.add(new_room)
else:
try:
rid_int = int(rid)
except ValueError:
print(f"⚠️ Invalid room ID format: {rid}")
continue
room = existing_by_id.get(rid_int)
if not room:
print(f"⚠️ No matching room in DB for ID: {rid_int}")
continue
if room.name != name:
room.name = name
if room.area_id != resolved_area_id:
room.area_id = resolved_area_id
if room.function_id != resolved_function_id:
room.function_id = resolved_function_id
for existing_id in existing_ids - seen_ids:
room = existing_by_id.get(existing_id)
if not room:
continue
# Skip if a newly added room matches this one — likely duplicate
if any(
r["name"] == room.name and
resolve_fk(r["area_id"], section_map, "section") == room.area_id and
resolve_fk(r["function_id"], function_map, "function") == room.function_id
for r in submitted_clean
if r.get("id") is None or str(r.get("id")).startswith("room-")
):
print(f"⚠️ Skipping deletion of likely duplicate: {room}")
continue
db.session.delete(room)
@classmethod
def validate_state(cls, submitted_items: list[dict]) -> list[str]:
errors = []
for index, item in enumerate(submitted_items):
label = f"Room #{index + 1}"
if not isinstance(item, dict):
errors.append(f"{label} is not a valid object.")
continue
name = item.get("name")
if not name or not str(name).strip():
errors.append(f"{label} is missing a name.")
raw_id = item.get("id")
if raw_id is not None:
try:
_ = int(raw_id)
except (ValueError, TypeError):
if not str(raw_id).startswith("room-"):
errors.append(f"{label} has an invalid ID: {raw_id}")
# These fields are FK IDs, so we're just checking for valid formats here.
for fk_field, fk_label in [("area_id", "Section"), ("function_id", "Function")]:
fk_val = item.get(fk_field)
if fk_val is None:
continue # Let the DB enforce nullability
try:
_ = int(fk_val)
except (ValueError, TypeError):
fk_val_str = str(fk_val)
if not fk_val_str.startswith("temp-"):
errors.append(f"{label} has invalid {fk_label} ID: {fk_val}")
return errors
Room.ui_eagerload = ( Room.ui_eagerload = (
joinedload(Room.area), joinedload(Room.area),
joinedload(Room.room_function), joinedload(Room.room_function),

View file

@ -112,7 +112,6 @@ def get_image_attachable_class_by_name(name: str):
return cls return cls
return None return None
def make_csv(export_func, columns, rows): def make_csv(export_func, columns, rows):
output = io.StringIO() output = io.StringIO()
writer = csv.writer(output) writer = csv.writer(output)

View file

@ -1,6 +1,5 @@
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from flask import current_app as app from flask import current_app as app
import re
from . import main from . import main

View file

@ -1,10 +1,5 @@
import io
import csv
import base64
import datetime import datetime
from flask import request, render_template, url_for, jsonify from flask import request, render_template, jsonify
from sqlalchemy.inspection import inspect
from . import main from . import main
from .helpers import FILTER_MAP, inventory_headers, worklog_headers, make_csv from .helpers import FILTER_MAP, inventory_headers, worklog_headers, make_csv

View file

@ -1,109 +1,15 @@
import json from flask import render_template
import traceback
from flask import request, flash, redirect, url_for, render_template, jsonify
from . import main from . import main
from .. import db from .. import db
from ..models import Brand, Item, Area, RoomFunction, Room, Image from ..models import Image
from ..utils.load import eager_load_room_relationships, chunk_list from ..utils.load import chunk_list
@main.route('/settings', methods=['GET', 'POST']) @main.route('/settings')
def settings(): def settings():
if request.method == 'POST':
form = request.form
try:
state = json.loads(form['formState'])
except Exception:
flash("Invalid form state submitted. JSON decode failed.", "danger")
traceback.print_exc()
return redirect(url_for('main.settings'))
try:
with db.session.begin():
# Sync each table and grab temp ID maps
brand_map = Brand.sync_from_state(state.get("brands", []))
type_map = Item.sync_from_state(state.get("types", []))
section_map = Area.sync_from_state(state.get("sections", []))
function_map = RoomFunction.sync_from_state(state.get("functions", []))
# Fix up room foreign keys based on real IDs
submitted_rooms = []
for room in state.get("rooms", []):
room = dict(room) # shallow copy
sid = room.get("area_id")
fid = room.get("function_id")
if sid is not None:
sid_key = str(sid)
if sid_key in section_map:
room["area_id"] = section_map[sid_key]
if fid is not None:
fid_key = str(fid)
if fid_key in function_map:
room["function_id"] = function_map[fid_key]
submitted_rooms.append(room)
Room.sync_from_state(
submitted_rooms=submitted_rooms,
section_map=section_map,
function_map=function_map
)
flash("Changes saved.", "success")
except Exception as e:
flash(f"Error saving changes: {e}", "danger")
return redirect(url_for('main.settings'))
# === GET ===
brands = db.session.query(Brand).order_by(Brand.name).all()
types = db.session.query(Item).order_by(Item.description).all()
sections = db.session.query(Area).order_by(Area.name).all()
functions = db.session.query(RoomFunction).order_by(RoomFunction.description).all()
rooms = eager_load_room_relationships(db.session.query(Room).order_by(Room.name)).all()
images = chunk_list(db.session.query(Image).order_by(Image.timestamp).all(), 6) images = chunk_list(db.session.query(Image).order_by(Image.timestamp).all(), 6)
return render_template('settings.html', return render_template('settings.html',
title="Settings", title="Settings",
brands=[b.serialize() for b in brands],
types=[{"id": t.id, "name": t.description} for t in types],
sections=[s.serialize() for s in sections],
functions=[f.serialize() for f in functions],
rooms=[r.serialize() for r in rooms],
image_list=images image_list=images
) )
@main.route("/api/settings", methods=["POST"])
def api_settings():
try:
payload = request.get_json(force=True)
except Exception as e:
return jsonify({"error": "Invalid JSON"}), 400
errors = []
errors += Brand.validate_state(payload.get("brands", []))
errors += Item.validate_state(payload.get("types", []))
errors += Area.validate_state(payload.get("sections", []))
errors += RoomFunction.validate_state(payload.get("functions", []))
errors += Room.validate_state(payload.get("rooms", []))
if errors:
return jsonify({"errors": errors}), 400
try:
with db.session.begin():
section_map = Area.sync_from_state(payload["sections"])
function_map = RoomFunction.sync_from_state(payload["functions"])
Brand.sync_from_state(payload["brands"])
Item.sync_from_state(payload["types"])
Room.sync_from_state(payload["rooms"], section_map, function_map)
except Exception as e:
db.session.rollback()
return jsonify({"errors": [str(e)]}), 500
return jsonify({"message": "Settings updated successfully."}), 200

View file

@ -1,6 +0,0 @@
def is_temp_id(val):
return (
val is None or
(isinstance(val, int) and val < 0) or
(isinstance(val, str) and val.startswith("temp-"))
)

View file

@ -1,28 +0,0 @@
from ..temp import is_temp_id
class ValidatableMixin:
VALIDATION_LABEL = "entry"
@classmethod
def validate_state(cls, submitted_items: list[dict]) -> list[str]:
errors = []
label = cls.VALIDATION_LABEL or cls.__name__
for index, item in enumerate(submitted_items):
if not isinstance(item, dict):
errors.append(f"{label.capitalize()} #{index + 1} is not a valid object.")
continue
name = item.get("name")
if not name or not str(name).strip():
errors.append(f"{label.capitalize()} #{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"{label.capitalize()} #{index + 1} has invalid ID: {raw_id}")
return errors