77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from sqlmodel import select
|
|
|
|
from src.db import get_db
|
|
from src.utils.decorators import auth_required
|
|
from src.resources.models import ResourceType, ResourceChange
|
|
from src.resources.schemas import (
|
|
ResourceTypeCreate, ResourceTypeResponse,
|
|
ResourceChangeCreate, ResourceChangeResponse
|
|
)
|
|
|
|
resources_router = APIRouter()
|
|
|
|
@auth_required()
|
|
@resources_router.post("/types", response_model=ResourceTypeResponse)
|
|
def create_resource_type(rt: ResourceTypeCreate, db: Session = Depends(get_db)):
|
|
db_rt = ResourceType.model_validate(rt)
|
|
db.add(db_rt)
|
|
db.commit()
|
|
db.refresh(db_rt)
|
|
return db_rt
|
|
|
|
|
|
@auth_required()
|
|
@resources_router.get("/types", response_model=List[ResourceTypeResponse])
|
|
def list_resource_types(db: Session = Depends(get_db)):
|
|
rts = db.execute(select(ResourceType)).scalars().all()
|
|
return rts
|
|
|
|
|
|
@auth_required()
|
|
@resources_router.get("/types/{type_id}", response_model=ResourceTypeResponse)
|
|
def get_resource_type(type_id: int, db: Session = Depends(get_db)):
|
|
rt = db.get(ResourceType, type_id)
|
|
if not rt:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Resource type not found")
|
|
|
|
return rt
|
|
|
|
|
|
@auth_required()
|
|
@resources_router.delete("/types/{type_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_resource_type(type_id: int, db: Session = Depends(get_db)):
|
|
rt = db.get(ResourceType, type_id)
|
|
if not rt:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Resource type not found")
|
|
db.delete(rt)
|
|
db.commit()
|
|
return
|
|
|
|
|
|
@auth_required()
|
|
@resources_router.post("/types/{type_id}/changes", response_model=ResourceChangeResponse)
|
|
def create_resource_change(type_id: int, change: ResourceChangeCreate, db: Session = Depends(get_db)):
|
|
rt = db.get(ResourceType, type_id)
|
|
if not rt:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Resource type not found")
|
|
|
|
db_change = ResourceChange.model_validate(change)
|
|
db_change.type = rt
|
|
db.add(db_change)
|
|
db.commit()
|
|
db.refresh(db_change)
|
|
return db_change
|
|
|
|
|
|
@auth_required()
|
|
@resources_router.get("/types/{type_id}/changes", response_model=List[ResourceChangeResponse])
|
|
def list_resource_changes(type_id: int, db: Session = Depends(get_db)):
|
|
rt = db.get(ResourceType, type_id)
|
|
if not rt:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Resource type not found")
|
|
return rt.changes
|