49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
"""Record routes."""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
|
|
from app.database import get_db
|
|
from app.models.record import Record
|
|
from app.schemas.record import RecordCreate, RecordUpdate, RecordResponse
|
|
|
|
router = APIRouter(prefix="/records", tags=["records"])
|
|
|
|
@router.get("/table/{table_id}", response_model=List[RecordResponse])
|
|
def list_records(table_id: int, db: Session = Depends(get_db)):
|
|
"""List all records for a table."""
|
|
return db.query(Record).filter(Record.column_id.in_(
|
|
db.query(Column.id).filter(Column.table_id == table_id)
|
|
)).all()
|
|
|
|
@router.post("", response_model=RecordResponse)
|
|
def create_record(record: RecordCreate, db: Session = Depends(get_db)):
|
|
"""Create new record."""
|
|
db_record = Record(**record.model_dump())
|
|
db.add(db_record)
|
|
db.commit()
|
|
db.refresh(db_record)
|
|
return db_record
|
|
|
|
@router.put("/{record_id}", response_model=RecordResponse)
|
|
def update_record(record_id: int, record: RecordUpdate, db: Session = Depends(get_db)):
|
|
"""Update record."""
|
|
db_record = db.query(Record).filter(Record.id == record_id).first()
|
|
if not db_record:
|
|
raise HTTPException(status_code=404, detail="Record not found")
|
|
for key, value in record.model_dump(exclude_unset=True).items():
|
|
setattr(db_record, key, value)
|
|
db.commit()
|
|
db.refresh(db_record)
|
|
return db_record
|
|
|
|
@router.delete("/{record_id}")
|
|
def delete_record(record_id: int, db: Session = Depends(get_db)):
|
|
"""Delete record."""
|
|
record = db.query(Record).filter(Record.id == record_id).first()
|
|
if not record:
|
|
raise HTTPException(status_code=404, detail="Record not found")
|
|
db.delete(record)
|
|
db.commit()
|
|
return {"message": "Record deleted"}
|