99fd82a025
Some checks failed
CI / backend-lint-and-test (push) Has been cancelled
- Remove IfcWallStandardCase from parser (covered by IfcWall as parent type) - Add seen_global_ids set to prevent duplicate element insertion - Use scalars().first() instead of scalar_one_or_none() for resilient lookup - Re-parse required: clears and rebuilds element data without duplicates
89 lines
2.8 KiB
Python
89 lines
2.8 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("/projects/{project_id}/elements/by-global-id/{global_id}", response_model=ElementDetailOut)
|
|
async def get_element_by_global_id(
|
|
project_id: UUID,
|
|
global_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Look up an element by its IFC GlobalId — used by the 3D viewer on click."""
|
|
query = (
|
|
select(Element)
|
|
.options(selectinload(Element.properties))
|
|
.where(Element.project_id == project_id)
|
|
.where(Element.global_id == global_id)
|
|
)
|
|
result = await db.execute(query)
|
|
element = result.scalars().first()
|
|
|
|
if not element:
|
|
raise HTTPException(status_code=404, detail="Element not found")
|
|
|
|
return element
|
|
|
|
@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
|