98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
import uuid
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.box import Box
|
|
from app.schemas.box import BoxCreate, BoxRead, BoxSummary, BoxUpdate
|
|
from app.schemas.object import ObjectSummary
|
|
from app.services.qrcodes import box_public_url
|
|
|
|
|
|
def _select_with_relations():
|
|
return select(Box).options(selectinload(Box.children), selectinload(Box.objects))
|
|
|
|
|
|
async def list_boxes(
|
|
db: AsyncSession, search: str | None = None, root_only: bool = False
|
|
) -> list[Box]:
|
|
stmt = select(Box).order_by(Box.name)
|
|
if root_only:
|
|
stmt = stmt.where(Box.parent_box_id.is_(None))
|
|
if search:
|
|
stmt = stmt.where(Box.name.ilike(f"%{search}%"))
|
|
result = await db.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_box(db: AsyncSession, box_id: uuid.UUID) -> Box | None:
|
|
stmt = _select_with_relations().where(Box.id == box_id)
|
|
result = await db.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def build_box_read(db: AsyncSession, box: Box) -> BoxRead:
|
|
ancestors: list[BoxSummary] = []
|
|
current_parent_id = box.parent_box_id
|
|
depth = 0
|
|
while current_parent_id is not None and depth < 100:
|
|
parent = await db.get(Box, current_parent_id)
|
|
if parent is None:
|
|
break
|
|
ancestors.append(BoxSummary(id=parent.id, name=parent.name))
|
|
current_parent_id = parent.parent_box_id
|
|
depth += 1
|
|
ancestors.reverse()
|
|
|
|
return BoxRead(
|
|
id=box.id,
|
|
name=box.name,
|
|
description=box.description,
|
|
parent_box_id=box.parent_box_id,
|
|
created_at=box.created_at,
|
|
updated_at=box.updated_at,
|
|
qr_code_url=box_public_url(box.id),
|
|
ancestors=ancestors,
|
|
children=[BoxSummary.model_validate(c) for c in box.children],
|
|
objects=[ObjectSummary.model_validate(o) for o in box.objects],
|
|
)
|
|
|
|
|
|
async def _is_descendant(db: AsyncSession, box_id: uuid.UUID, candidate_ancestor_id: uuid.UUID) -> bool:
|
|
"""Whether candidate_ancestor_id is box_id itself or one of its descendants."""
|
|
if box_id == candidate_ancestor_id:
|
|
return True
|
|
stmt = select(Box.id).where(Box.parent_box_id == box_id)
|
|
result = await db.execute(stmt)
|
|
for child_id in result.scalars().all():
|
|
if await _is_descendant(db, child_id, candidate_ancestor_id):
|
|
return True
|
|
return False
|
|
|
|
|
|
async def create_box(db: AsyncSession, data: BoxCreate) -> Box:
|
|
box = Box(**data.model_dump())
|
|
db.add(box)
|
|
await db.commit()
|
|
return await get_box(db, box.id)
|
|
|
|
|
|
async def update_box(db: AsyncSession, box: Box, data: BoxUpdate) -> Box:
|
|
updates = data.model_dump(exclude_unset=True)
|
|
new_parent_id = updates.get("parent_box_id", box.parent_box_id)
|
|
if new_parent_id is not None and await _is_descendant(db, box.id, new_parent_id):
|
|
raise ValueError("A box cannot be moved into itself or one of its own descendants")
|
|
for field, value in updates.items():
|
|
setattr(box, field, value)
|
|
await db.commit()
|
|
return await get_box(db, box.id)
|
|
|
|
|
|
async def delete_box(db: AsyncSession, box: Box) -> None:
|
|
child_count = await db.scalar(select(Box.id).where(Box.parent_box_id == box.id).limit(1))
|
|
if child_count is not None:
|
|
raise ValueError("Cannot delete a box that still contains nested boxes")
|
|
await db.delete(box)
|
|
await db.commit()
|