Add inventory management templates and utility functions

- Created HTML templates for inventory index, layout, search, settings, table, user, and worklog.
- Implemented utility functions for eager loading relationships in SQLAlchemy.
- Added validation mixin for form submissions.
- Updated project configuration files (pyproject.toml and setup.cfg) for package management.
This commit is contained in:
Yaro Kasear 2025-07-08 11:47:22 -05:00
parent 602bb77e22
commit 9803db17ab
51 changed files with 76 additions and 16 deletions

View file

@ -0,0 +1,24 @@
from typing import TYPE_CHECKING
from inventory import db # Yes, this works when run from project root with Alembic
from .areas import Area
from .brands import Brand
from .items import Item
from .inventory import Inventory
from .room_functions import RoomFunction
from .users import User
from .work_log import WorkLog
from .rooms import Room
__all__ = [
"db",
"Area",
"Brand",
"Item",
"Inventory",
"RoomFunction",
"User",
"WorkLog",
"Room",
]

123
inventory/models/areas.py Normal file
View file

@ -0,0 +1,123 @@
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__ = 'Areas'
VALIDATION_LABEL = "Area"
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
name: Mapped[Optional[str]] = mapped_column("Area", 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())
print(f"Existing area IDs: {existing_ids}")
print(f"Submitted area IDs: {seen_ids}")
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
print(f" Adding area: {submitted_name}")
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:
print(f"✏️ Updating area {area.id}: '{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)
print(f"🗑️ Removing area: {area.name}")
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

115
inventory/models/brands.py Normal file
View file

@ -0,0 +1,115 @@
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 Brand(ValidatableMixin, db.Model):
__tablename__ = 'Brands'
VALIDATION_LABEL = 'Brand'
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
name: Mapped[str] = mapped_column("Brand", Unicode(255), nullable=False)
inventory: Mapped[List['Inventory']] = relationship('Inventory', back_populates='brand')
def __init__(self, name: str):
self.name = name
def __repr__(self):
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())
print(f"Existing brand IDs: {existing_ids}")
print(f"Submitted brand IDs: {seen_ids}")
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
print(f" Adding brand: {name}")
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:
print(f"✏️ Updating brand {obj.id}: '{obj.name}''{name}'")
obj.name = name
for id_to_remove in existing_ids - seen_ids:
db.session.delete(existing_by_id[id_to_remove])
print(f"🗑️ Removing brand 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

View file

@ -0,0 +1,128 @@
from typing import Any, List, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from .brands import Brand
from .items import Item
from .work_log import WorkLog
from .rooms import Room
from sqlalchemy import Boolean, ForeignKey, Identity, Index, Integer, Unicode, DateTime, text
from sqlalchemy.orm import Mapped, mapped_column, relationship
import datetime
from . import db
class Inventory(db.Model):
__tablename__ = 'Inventory'
__table_args__ = (
Index('Inventory$Bar Code', 'Bar Code'),
)
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
timestamp: Mapped[datetime.datetime] = mapped_column('Date Entered', DateTime)
condition: Mapped[str] = mapped_column('Working Condition', Unicode(255))
needed: Mapped[str] = mapped_column("Needed", Unicode(255))
type_id: Mapped[Optional[int]] = mapped_column('Item Type', Integer, ForeignKey("Items.ID"), nullable=True)
inventory_name: Mapped[Optional[str]] = mapped_column('Inventory #', Unicode(255))
serial: Mapped[Optional[str]] = mapped_column('Serial #', Unicode(255))
model: Mapped[Optional[str]] = mapped_column('Model #', Unicode(255))
notes: Mapped[Optional[str]] = mapped_column('Notes', Unicode(255))
owner_id = mapped_column('Owner', Integer, ForeignKey('Users.ID'))
brand_id: Mapped[Optional[int]] = mapped_column("Brand", Integer, ForeignKey("Brands.ID"))
# Photo: Mapped[Optional[str]] = mapped_column(String(8000)) Will be replacing with something that actually works.
location_id: Mapped[Optional[str]] = mapped_column(ForeignKey("Rooms.ID"))
barcode: Mapped[Optional[str]] = mapped_column('Bar Code', Unicode(255))
shared: Mapped[Optional[bool]] = mapped_column(Boolean, server_default=text('((0))'))
location: Mapped[Optional['Room']] = relationship('Room', back_populates='inventory')
owner = relationship('User', back_populates='inventory')
brand: Mapped[Optional['Brand']] = relationship('Brand', back_populates='inventory')
item: Mapped['Item'] = relationship('Item', back_populates='inventory')
work_logs: Mapped[List['WorkLog']] = relationship('WorkLog', back_populates='work_item')
def __init__(self, timestamp: datetime.datetime, condition: str, needed: str, type_id: Optional[int] = None,
inventory_name: Optional[str] = None, serial: Optional[str] = None,
model: Optional[str] = None, notes: Optional[str] = None, owner_id: Optional[int] = None,
brand_id: Optional[int] = None, location_id: Optional[str] = None, barcode: Optional[str] = None,
shared: bool = False):
self.timestamp = timestamp
self.condition = condition
self.needed = needed
self.type_id = type_id
self.inventory_name = inventory_name
self.serial = serial
self.model = model
self.notes = notes
self.owner_id = owner_id
self.brand_id = brand_id
self.location_id = location_id
self.barcode = barcode
self.shared = shared
def __repr__(self):
parts = [f"id={self.id}"]
if self.inventory_name:
parts.append(f"name={repr(self.inventory_name)}")
if self.item:
parts.append(f"item={repr(self.item.description)}")
if self.notes:
parts.append(f"notes={repr(self.notes)}")
if self.owner:
parts.append(f"owner={repr(self.owner.full_name)}")
if self.location:
parts.append(f"location={repr(self.location.full_name)}")
return f"<Inventory({', '.join(parts)})>"
@property
def identifier(self) -> str:
if self.inventory_name:
return f"Name: {self.inventory_name}"
elif self.barcode:
return f"Bar: {self.barcode}"
elif self.serial:
return f"Serial: {self.serial}"
else:
return f"ID: {self.id}"
def serialize(self) -> dict[str, Any]:
return {
'id': self.id,
'timestamp': self.timestamp.isoformat() if self.timestamp else None,
'condition': self.condition,
'needed': self.needed,
'type_id': self.type_id,
'inventory_name': self.inventory_name,
'serial': self.serial,
'model': self.model,
'notes': self.notes,
'owner_id': self.owner_id,
'brand_id': self.brand_id,
'location_id': self.location_id,
'barcode': self.barcode,
'shared': self.shared
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Inventory":
timestamp_str = data.get("timestamp")
return cls(
timestamp = datetime.datetime.fromisoformat(str(timestamp_str)) if timestamp_str else datetime.datetime.now(),
condition=data.get("condition", "Unverified"),
needed=data.get("needed", ""),
type_id=data["type_id"],
inventory_name=data.get("inventory_name"),
serial=data.get("serial"),
model=data.get("model"),
notes=data.get("notes"),
owner_id=data.get("owner_id"),
brand_id=data.get("brand_id"),
location_id=data.get("location_id"),
barcode=data.get("barcode"),
shared=bool(data.get("shared", False))
)

116
inventory/models/items.py Normal file
View file

@ -0,0 +1,116 @@
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

View file

@ -0,0 +1,90 @@
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 RoomFunction(ValidatableMixin, db.Model):
__tablename__ = 'Room Functions'
VALIDATION_LABEL = "Function"
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
description: Mapped[Optional[str]] = mapped_column("Function", Unicode(255), nullable=True)
rooms: Mapped[List['Room']] = relationship('Room', back_populates='room_function')
def __init__(self, description: Optional[str] = None):
self.description = description
def __repr__(self):
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())
print(f"Existing function IDs: {existing_ids}")
print(f"Submitted function IDs: {seen_ids}")
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
print(f" Adding function: {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 function {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 function 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

199
inventory/models/rooms.py Normal file
View file

@ -0,0 +1,199 @@
from typing import Optional, TYPE_CHECKING, List
if TYPE_CHECKING:
from .areas import Area
from .room_functions import RoomFunction
from .inventory import Inventory
from .users import User
from sqlalchemy import ForeignKey, Identity, Integer, Unicode
from sqlalchemy.orm import Mapped, mapped_column, relationship
from . import db
from ..utils.validation import ValidatableMixin
class Room(ValidatableMixin, db.Model):
__tablename__ = 'Rooms'
VALIDATION_LABEL = "Room"
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
name: Mapped[Optional[str]] = mapped_column("Room", Unicode(255), nullable=True)
area_id: Mapped[Optional[int]] = mapped_column("Area", Integer, ForeignKey("Areas.ID"))
function_id: Mapped[Optional[int]] = mapped_column("Function", Integer, ForeignKey("Room Functions.ID"))
area: Mapped[Optional['Area']] = relationship('Area', back_populates='rooms')
room_function: Mapped[Optional['RoomFunction']] = relationship('RoomFunction', back_populates='rooms')
inventory: Mapped[List['Inventory']] = relationship('Inventory', back_populates='location')
users: Mapped[List['User']] = relationship('User', back_populates='location')
def __init__(self, name: Optional[str] = None, area_id: Optional[int] = None, function_id: Optional[int] = None):
self.name = name
self.area_id = area_id
self.function_id = function_id
def __repr__(self):
return f"<Room(id={self.id}, room={repr(self.name)}, area_id={self.area_id}, function_id={self.function_id})>"
@property
def full_name(self):
name = self.name or ""
func = self.room_function.description if self.room_function else ""
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):
print(f"Resolving {label} ID: {key} using map: {fk_map}")
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")
section_id = room.get("section_id")
function_id = room.get("function_id")
submitted_clean.append({
"id": rid,
"name": name,
"section_id": section_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())
print(f"Existing room IDs: {existing_ids}")
print(f"Submitted room IDs: {seen_ids}")
for entry in submitted_clean:
rid = entry.get("id")
name = entry["name"]
resolved_section_id = resolve_fk(entry.get("section_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_section_id, function_id=resolved_function_id)
db.session.add(new_room)
print(f" Adding room: {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
updated = False
if room.name != name:
print(f"✏️ Updating room name {room.id}: '{room.name}''{name}'")
room.name = name
updated = True
if room.area_id != resolved_section_id:
print(f"✏️ Updating room area {room.id}: {room.area_id}{resolved_section_id}")
room.area_id = resolved_section_id
updated = True
if room.function_id != resolved_function_id:
print(f"✏️ Updating room function {room.id}: {room.function_id}{resolved_function_id}")
room.function_id = resolved_function_id
updated = True
if not updated:
print(f"✅ No changes to room {room.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["section_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)
print(f"🗑️ Removing room: {room}")
@classmethod
def validate_state(cls, submitted_items: list[dict]) -> list[str]:
print("VALIDATING")
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 [("section_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

68
inventory/models/users.py Normal file
View file

@ -0,0 +1,68 @@
from typing import Any, List, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from .inventory import Inventory
from .rooms import Room
from .work_log import WorkLog
from sqlalchemy import Boolean, ForeignKey, Identity, Integer, Unicode, text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from . import db
class User(db.Model):
__tablename__ = 'Users'
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
staff: Mapped[Optional[bool]] = mapped_column("Staff", Boolean, server_default=text('((0))'))
active: Mapped[Optional[bool]] = mapped_column("Active", Boolean, server_default=text('((0))'))
last_name: Mapped[Optional[str]] = mapped_column("Last", Unicode(255), nullable=True)
first_name: Mapped[Optional[str]] = mapped_column("First", Unicode(255), nullable=True)
location_id: Mapped[Optional[int]] = mapped_column(ForeignKey("Rooms.ID"), nullable=True)
supervisor_id: Mapped[Optional[int]] = mapped_column("Supervisor", Integer, ForeignKey("Users.ID"))
supervisor: Mapped[Optional['User']] = relationship('User', remote_side='User.id', back_populates='subordinates')
subordinates: Mapped[List['User']] = relationship('User', back_populates='supervisor')
work_logs: Mapped[List['WorkLog']] = relationship('WorkLog', back_populates='contact')
location: Mapped[Optional['Room']] = relationship('Room', back_populates='users')
inventory: Mapped[List['Inventory']] = relationship('Inventory', back_populates='owner')
@property
def full_name(self) -> str:
return f"{self.first_name or ''} {self.last_name or ''}".strip()
def __init__(self, first_name: Optional[str] = None, last_name: Optional[str] = None,
location_id: Optional[int] = None, supervisor_id: Optional[int] = None,
staff: Optional[bool] = False, active: Optional[bool] = False):
self.first_name = first_name
self.last_name = last_name
self.location_id = location_id
self.supervisor_id = supervisor_id
self.staff = staff
self.active = active
def __repr__(self):
return f"<User(id={self.id}, first_name={repr(self.first_name)}, last_name={repr(self.last_name)}, " \
f"location={repr(self.location)}, staff={self.staff}, active={self.active})>"
def serialize(self):
return {
'id': self.id,
'first_name': self.first_name,
'last_name': self.last_name,
'location_id': self.location_id,
'supervisor_id': self.supervisor_id,
'staff': self.staff,
'active': self.active
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "User":
return cls(
staff=bool(data.get("staff", False)),
active=bool(data.get("active", False)),
last_name=data.get("last_name"),
first_name=data.get("first_name"),
location_id=data.get("location_id"),
supervisor_id=data.get("supervisor_id")
)

View file

@ -0,0 +1,76 @@
from typing import Optional, Any, TYPE_CHECKING
if TYPE_CHECKING:
from .inventory import Inventory
from .users import User
from sqlalchemy import Boolean, ForeignKeyConstraint, Identity, Integer, ForeignKey, Unicode, DateTime, text
from sqlalchemy.orm import Mapped, mapped_column, relationship
import datetime
from . import db
class WorkLog(db.Model):
__tablename__ = 'Work Log'
__table_args__ = (
ForeignKeyConstraint(['Work Item'], ['Inventory.ID'], name='Work Log$InventoryWork Log'),
)
id: Mapped[int] = mapped_column("ID", Integer, Identity(start=1, increment=1), primary_key=True)
start_time: Mapped[Optional[datetime.datetime]] = mapped_column('Start Timestamp', DateTime)
end_time: Mapped[Optional[datetime.datetime]] = mapped_column('End Timestamp', DateTime)
notes: Mapped[Optional[str]] = mapped_column('Description & Notes', Unicode())
complete: Mapped[Optional[bool]] = mapped_column("Complete", Boolean, server_default=text('((0))'))
followup: Mapped[Optional[bool]] = mapped_column('Needs Follow-Up', Boolean, server_default=text('((0))'))
contact_id: Mapped[Optional[int]] = mapped_column('Point of Contact', Integer, ForeignKey("Users.ID"))
analysis: Mapped[Optional[bool]] = mapped_column('Needs Quick Analysis', Boolean, server_default=text('((0))'))
work_item_id: Mapped[Optional[int]] = mapped_column("Work Item", Integer, ForeignKey("Inventory.ID"))
work_item: Mapped[Optional['Inventory']] = relationship('Inventory', back_populates='work_logs')
contact: Mapped[Optional['User']] = relationship('User', back_populates='work_logs')
def __init__(self, start_time: Optional[datetime.datetime] = None, end_time: Optional[datetime.datetime] = None,
notes: Optional[str] = None, complete: Optional[bool] = False,
followup: Optional[bool] = False, contact_id: Optional[int] = None,
analysis: Optional[bool] = False, work_item_id: Optional[int] = None):
self.start_time = start_time
self.end_time = end_time
self.notes = notes
self.complete = complete
self.followup = followup
self.contact_id = contact_id
self.analysis = analysis
self.work_item_id = work_item_id
def __repr__(self):
return f"<WorkLog(id={self.id}, start_time={self.start_time}, end_time={self.end_time}, " \
f"notes={repr(self.notes)}, complete={self.complete}, followup={self.followup}, " \
f"contact_id={self.contact_id}, analysis={self.analysis}, work_item_id={self.work_item_id})>"
def serialize(self):
return {
'id': self.id,
'start_time': self.start_time.isoformat() if self.start_time else None,
'end_time': self.end_time.isoformat() if self.end_time else None,
'notes': self.notes,
'complete': self.complete,
'followup': self.followup,
'contact_id': self.contact_id,
'analysis': self.analysis,
'work_item_id': self.work_item_id
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "WorkLog":
start_time_str = data.get("start_time")
end_time_str = data.get("end_time")
return cls(
start_time=datetime.datetime.fromisoformat(str(start_time_str)) if start_time_str else datetime.datetime.now(),
end_time=datetime.datetime.fromisoformat(str(end_time_str)) if end_time_str else None,
notes=data.get("notes"),
complete=bool(data.get("complete", False)),
followup=bool(data.get("followup", False)),
analysis=bool(data.get("analysis", False)),
contact_id=data.get("contact_id"),
work_item_id=data.get("work_item_id")
)