Agent Zero
111 lines
3.2 KiB
Python
111 lines
3.2 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
|
|
from app.database import get_db
|
|
from app.models.user import User
|
|
from app.models.view import View
|
|
from app.schemas.view import ViewCreate, ViewRead, ViewUpdate
|
|
from app.routers.auth import get_current_user
|
|
from app.routers.tables import check_table_access
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/tables/{table_id}/views", response_model=List[ViewRead])
|
|
def list_views(
|
|
table_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""List views for table."""
|
|
check_table_access(table_id, db, current_user)
|
|
return db.query(View).filter(View.table_id == table_id).all()
|
|
|
|
|
|
@router.post("/tables/{table_id}/views", response_model=ViewRead, status_code=201)
|
|
def create_view(
|
|
table_id: int,
|
|
data: ViewCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Create view."""
|
|
check_table_access(table_id, db, current_user)
|
|
|
|
# If is_default, unset other defaults
|
|
if data.is_default:
|
|
db.query(View).filter(
|
|
View.table_id == table_id, View.is_default == True
|
|
).update({"is_default": False})
|
|
|
|
view = View(
|
|
table_id=table_id,
|
|
name=data.name,
|
|
type=data.type,
|
|
config=data.config,
|
|
is_default=data.is_default,
|
|
created_by=current_user.id,
|
|
)
|
|
db.add(view)
|
|
db.commit()
|
|
db.refresh(view)
|
|
return view
|
|
|
|
|
|
@router.get("/views/{view_id}", response_model=ViewRead)
|
|
def get_view(
|
|
view_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get view config."""
|
|
view = db.query(View).filter(View.id == view_id).first()
|
|
if not view:
|
|
raise HTTPException(status_code=404, detail="View not found")
|
|
check_table_access(view.table_id, db, current_user)
|
|
return view
|
|
|
|
|
|
@router.put("/views/{view_id}", response_model=ViewRead)
|
|
def update_view(
|
|
view_id: int,
|
|
data: ViewUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Update view."""
|
|
view = db.query(View).filter(View.id == view_id).first()
|
|
if not view:
|
|
raise HTTPException(status_code=404, detail="View not found")
|
|
check_table_access(view.table_id, db, current_user)
|
|
|
|
# If setting as default, unset other defaults
|
|
if data.is_default and not view.is_default:
|
|
db.query(View).filter(
|
|
View.table_id == view.table_id, View.is_default == True
|
|
).update({"is_default": False})
|
|
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(view, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(view)
|
|
return view
|
|
|
|
|
|
@router.delete("/views/{view_id}")
|
|
def delete_view(
|
|
view_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Delete view."""
|
|
view = db.query(View).filter(View.id == view_id).first()
|
|
if not view:
|
|
raise HTTPException(status_code=404, detail="View not found")
|
|
check_table_access(view.table_id, db, current_user)
|
|
db.delete(view)
|
|
db.commit()
|
|
return {"message": "View deleted"}
|