Jonas
4c9deda66a
Add the core labeling data model (label sets, ad-hoc Main->Sub class hierarchy, frames, bounding-box labels, per-frame/per-set label status) behind a FastAPI app, with SQLite as the default swappable DATABASE_URL. Multi-user support is attribution-only for now (get-or-create by username, no login flow yet). Dataset versioning/promotion is intentionally deferred - it needs its own design pass around snapshot semantics. Each test gets a fully isolated app+DB via create_app(database_url=...) rather than relying on process-global state. 13/13 tests pass; also verified live end-to-end against a running uvicorn instance. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
"""Bounding-box labels and per-frame/per-set label status."""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import models, schemas
|
|
from ..db import get_db
|
|
from ..users import get_or_create_user
|
|
|
|
router = APIRouter(tags=["labels"])
|
|
|
|
|
|
def _require_frame_and_set(db: Session, frame_id: int, set_id: int) -> None:
|
|
if db.get(models.Frame, frame_id) is None:
|
|
raise HTTPException(404, "frame not found")
|
|
if db.get(models.LabelSet, set_id) is None:
|
|
raise HTTPException(404, "set not found")
|
|
|
|
|
|
@router.post("/frames/{frame_id}/sets/{set_id}/labels", response_model=schemas.LabelRead, status_code=201)
|
|
def create_label(frame_id: int, set_id: int, payload: schemas.LabelCreate, db: Session = Depends(get_db)):
|
|
_require_frame_and_set(db, frame_id, set_id)
|
|
|
|
sub_class = db.get(models.SubClass, payload.sub_class_id)
|
|
if sub_class is None or sub_class.main_class.set_id != set_id:
|
|
raise HTTPException(422, "sub_class_id does not belong to this set")
|
|
|
|
label = models.Label(
|
|
frame_id=frame_id,
|
|
set_id=set_id,
|
|
sub_class_id=payload.sub_class_id,
|
|
x=payload.x,
|
|
y=payload.y,
|
|
width=payload.width,
|
|
height=payload.height,
|
|
source=payload.source,
|
|
created_by_id=get_or_create_user(db, payload.created_by).id if payload.created_by else None,
|
|
)
|
|
db.add(label)
|
|
db.commit()
|
|
db.refresh(label)
|
|
return label
|
|
|
|
|
|
@router.get("/frames/{frame_id}/sets/{set_id}/labels", response_model=list[schemas.LabelRead])
|
|
def list_labels(frame_id: int, set_id: int, db: Session = Depends(get_db)):
|
|
_require_frame_and_set(db, frame_id, set_id)
|
|
return db.query(models.Label).filter_by(frame_id=frame_id, set_id=set_id).all()
|
|
|
|
|
|
@router.patch("/labels/{label_id}", response_model=schemas.LabelRead)
|
|
def update_label(label_id: int, payload: schemas.LabelUpdate, db: Session = Depends(get_db)):
|
|
label = db.get(models.Label, label_id)
|
|
if label is None:
|
|
raise HTTPException(404, "label not found")
|
|
for field, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(label, field, value)
|
|
db.commit()
|
|
db.refresh(label)
|
|
return label
|
|
|
|
|
|
@router.delete("/labels/{label_id}", status_code=204)
|
|
def delete_label(label_id: int, db: Session = Depends(get_db)):
|
|
label = db.get(models.Label, label_id)
|
|
if label is None:
|
|
raise HTTPException(404, "label not found")
|
|
db.delete(label)
|
|
db.commit()
|
|
|
|
|
|
@router.get("/frames/{frame_id}/sets/{set_id}/status", response_model=schemas.FrameSetStatusRead)
|
|
def get_status(frame_id: int, set_id: int, db: Session = Depends(get_db)):
|
|
_require_frame_and_set(db, frame_id, set_id)
|
|
status = db.query(models.FrameSetStatus).filter_by(frame_id=frame_id, set_id=set_id).first()
|
|
if status is None:
|
|
status = models.FrameSetStatus(frame_id=frame_id, set_id=set_id, status=models.LabelStatus.UNLABELED)
|
|
db.add(status)
|
|
db.commit()
|
|
db.refresh(status)
|
|
return status
|
|
|
|
|
|
@router.put("/frames/{frame_id}/sets/{set_id}/status", response_model=schemas.FrameSetStatusRead)
|
|
def set_status(frame_id: int, set_id: int, payload: schemas.FrameSetStatusUpdate, db: Session = Depends(get_db)):
|
|
_require_frame_and_set(db, frame_id, set_id)
|
|
status = db.query(models.FrameSetStatus).filter_by(frame_id=frame_id, set_id=set_id).first()
|
|
if status is None:
|
|
status = models.FrameSetStatus(frame_id=frame_id, set_id=set_id)
|
|
db.add(status)
|
|
status.status = payload.status
|
|
if payload.updated_by:
|
|
status.updated_by_id = get_or_create_user(db, payload.updated_by).id
|
|
db.commit()
|
|
db.refresh(status)
|
|
return status
|