46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
from sqlalchemy.orm import joinedload, selectinload
|
|
from .models import User, Room, Inventory, WorkLog
|
|
from . import db
|
|
|
|
def eager_load_user_relationships(query):
|
|
return query.options(
|
|
joinedload(User.supervisor),
|
|
joinedload(User.location).joinedload(Room.room_function)
|
|
)
|
|
|
|
def eager_load_inventory_relationships(query):
|
|
return query.options(
|
|
joinedload(Inventory.owner),
|
|
joinedload(Inventory.brand),
|
|
joinedload(Inventory.item),
|
|
selectinload(Inventory.location).selectinload(Room.room_function)
|
|
)
|
|
|
|
def eager_load_room_relationships(query):
|
|
return query.options(
|
|
joinedload(Room.area),
|
|
joinedload(Room.room_function),
|
|
selectinload(Room.inventory),
|
|
selectinload(Room.users)
|
|
)
|
|
|
|
def eager_load_worklog_relationships(query):
|
|
return query.options(
|
|
joinedload(WorkLog.contact),
|
|
joinedload(WorkLog.work_item)
|
|
)
|
|
|
|
def chunk_list(lst, chunk_size):
|
|
return [lst[i:i + chunk_size] for i in range(0, len(lst), chunk_size)]
|
|
|
|
def add_named_entities(items: list[str], model, attr: str, mapper: dict | None = None):
|
|
for name in items:
|
|
clean = name.strip()
|
|
if clean:
|
|
print(f"Creating new {attr}: {clean}")
|
|
new_obj = model(**{attr: clean})
|
|
db.session.add(new_obj)
|
|
if mapper is not None:
|
|
db.session.flush()
|
|
mapper[clean] = new_obj.id
|
|
print(f"New {attr} '{clean}' added with ID {new_obj.id}")
|