69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
"""API routes for querying building elements."""
|
|
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.database import get_db
|
|
from app.models.element import Element, Project
|
|
from app.schemas.element import ElementDetailOut, ElementOut, ProjectOut
|
|
|
|
router = APIRouter(tags=["elements"])
|
|
|
|
|
|
@router.get("/projects", response_model=list[ProjectOut])
|
|
async def list_projects(db: AsyncSession = Depends(get_db)):
|
|
"""List all uploaded IFC projects."""
|
|
query = select(Project)
|
|
result = await db.execute(query)
|
|
projects = result.scalars().all()
|
|
|
|
response = []
|
|
for project in projects:
|
|
count_query = select(func.count()).where(Element.project_id == project.id)
|
|
count_result = await db.execute(count_query)
|
|
count = count_result.scalar()
|
|
out = ProjectOut.model_validate(project)
|
|
out.element_count = count
|
|
response.append(out)
|
|
|
|
return response
|
|
|
|
|
|
@router.get("/projects/{project_id}/elements", response_model=list[ElementOut])
|
|
async def list_elements(
|
|
project_id: UUID,
|
|
ifc_type: str | None = None,
|
|
storey: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List elements for a project, optionally filtered by type or storey."""
|
|
query = select(Element).where(Element.project_id == project_id)
|
|
|
|
if ifc_type:
|
|
query = query.where(Element.ifc_type == ifc_type)
|
|
if storey:
|
|
query = query.where(Element.storey == storey)
|
|
|
|
result = await db.execute(query)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.get("/elements/{element_id}", response_model=ElementDetailOut)
|
|
async def get_element(element_id: UUID, db: AsyncSession = Depends(get_db)):
|
|
"""Get a single element with all its properties."""
|
|
query = (
|
|
select(Element)
|
|
.options(selectinload(Element.properties))
|
|
.where(Element.id == element_id)
|
|
)
|
|
result = await db.execute(query)
|
|
element = result.scalar_one_or_none()
|
|
|
|
if not element:
|
|
raise HTTPException(status_code=404, detail="Element not found")
|
|
|
|
return element
|