149 lines
3.8 KiB
Python
149 lines
3.8 KiB
Python
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
from sqlmodel import func, select
|
|
|
|
from app.api.deps import CurrentUser, SessionDep
|
|
from app.models.base import (
|
|
ApiTags,
|
|
Message,
|
|
RowId,
|
|
)
|
|
from app.models.route import (
|
|
Route,
|
|
RouteCreate,
|
|
RouteUpdate,
|
|
RoutePublic,
|
|
RoutesPublic,
|
|
)
|
|
from app.models.user import (
|
|
PermissionModule,
|
|
PermissionPart,
|
|
PermissionRight,
|
|
)
|
|
|
|
router = APIRouter(prefix="/routes", tags=[ApiTags.ROUTES])
|
|
|
|
|
|
# region # Routes ########################################################
|
|
|
|
|
|
@router.get("/", response_model=RoutesPublic)
|
|
def read_routes(
|
|
session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
|
|
) -> Any:
|
|
"""
|
|
Retrieve all routes.
|
|
"""
|
|
|
|
if current_user.has_permissions(
|
|
module=PermissionModule.ROUTE,
|
|
part=PermissionPart.ADMIN,
|
|
rights=PermissionRight.READ,
|
|
):
|
|
count_statement = select(func.count()).select_from(Route)
|
|
count = session.exec(count_statement).one()
|
|
statement = select(Route).offset(skip).limit(limit)
|
|
routes = session.exec(statement).all()
|
|
return RoutesPublic(data=routes, count=count)
|
|
|
|
return RoutesPublic(data=[], count=0)
|
|
|
|
|
|
@router.get("/{id}", response_model=RoutePublic)
|
|
def read_route(session: SessionDep, current_user: CurrentUser, id: RowId) -> Any:
|
|
"""
|
|
Get route by ID.
|
|
"""
|
|
route = session.get(Route, id)
|
|
if not route:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Route not found"
|
|
)
|
|
|
|
if not current_user.has_permissions(
|
|
module=PermissionModule.ROUTE,
|
|
part=PermissionPart.ADMIN,
|
|
rights=PermissionRight.READ,
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions"
|
|
)
|
|
|
|
return route
|
|
|
|
|
|
@router.post("/", response_model=RoutePublic)
|
|
def create_route(
|
|
*, session: SessionDep, current_user: CurrentUser, route_in: RouteCreate
|
|
) -> Any:
|
|
"""
|
|
Create new route.
|
|
"""
|
|
|
|
if not current_user.has_permissions(
|
|
module=PermissionModule.ROUTE,
|
|
part=PermissionPart.ADMIN,
|
|
rights=PermissionRight.CREATE,
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions"
|
|
)
|
|
|
|
route = Route.create(create_obj=route_in, session=session)
|
|
return route
|
|
|
|
|
|
@router.put("/{id}", response_model=RoutePublic)
|
|
def update_route(
|
|
*, session: SessionDep, current_user: CurrentUser, id: RowId, route_in: RouteUpdate
|
|
) -> Any:
|
|
"""
|
|
Update a route.
|
|
"""
|
|
route = session.get(Route, id)
|
|
if not route:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Route not found"
|
|
)
|
|
|
|
if not current_user.has_permissions(
|
|
module=PermissionModule.ROUTE,
|
|
part=PermissionPart.ADMIN,
|
|
rights=PermissionRight.UPDATE,
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions"
|
|
)
|
|
|
|
route = Route.update(db_obj=route, in_obj=route_in, session=session)
|
|
return route
|
|
|
|
|
|
@router.delete("/{id}")
|
|
def delete_route(session: SessionDep, current_user: CurrentUser, id: RowId) -> Message:
|
|
"""
|
|
Delete a route.
|
|
"""
|
|
route = session.get(Route, id)
|
|
if not route:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Route not found"
|
|
)
|
|
|
|
if not current_user.has_permissions(
|
|
module=PermissionModule.ROUTE,
|
|
part=PermissionPart.ADMIN,
|
|
rights=PermissionRight.DELETE,
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions"
|
|
)
|
|
|
|
session.delete(route)
|
|
session.commit()
|
|
return Message(message="Route deleted successfully")
|
|
|
|
|
|
# endregion
|