93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.crud import audit as audit_crud
|
|
from app.crud import static_connection as static_connection_crud
|
|
from app.deps import require_permission
|
|
from app.models.user import User
|
|
from app.schemas.static_connection import (
|
|
StaticConnectionCreate,
|
|
StaticConnectionRead,
|
|
StaticConnectionUpdate,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=list[StaticConnectionRead])
|
|
async def list_static_connections(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_permission("static_connections:read")),
|
|
):
|
|
return await static_connection_crud.list_all(db, skip=skip, limit=limit)
|
|
|
|
|
|
@router.post("", response_model=StaticConnectionRead, status_code=status.HTTP_201_CREATED)
|
|
async def create_static_connection(
|
|
payload: StaticConnectionCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
actor: User = Depends(require_permission("static_connections:write")),
|
|
):
|
|
sc = await static_connection_crud.create(db, od_mm=payload.od_mm, od_in=payload.od_in)
|
|
await audit_crud.log(
|
|
db, actor=actor, action="create", resource_type="StaticConnection",
|
|
resource_id=sc.id,
|
|
description=audit_crud.describe_create("StaticConnection", sc.id, f"{sc.od_mm} mm"),
|
|
)
|
|
return sc
|
|
|
|
|
|
@router.get("/{static_connection_id}", response_model=StaticConnectionRead)
|
|
async def get_static_connection(
|
|
static_connection_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_permission("static_connections:read")),
|
|
):
|
|
sc = await static_connection_crud.get(db, static_connection_id)
|
|
if sc is None:
|
|
raise HTTPException(status_code=404, detail="Static connection not found")
|
|
return sc
|
|
|
|
|
|
@router.patch("/{static_connection_id}", response_model=StaticConnectionRead)
|
|
async def update_static_connection(
|
|
static_connection_id: int,
|
|
payload: StaticConnectionUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
actor: User = Depends(require_permission("static_connections:write")),
|
|
):
|
|
sc = await static_connection_crud.get(db, static_connection_id)
|
|
if sc is None:
|
|
raise HTTPException(status_code=404, detail="Static connection not found")
|
|
changes = payload.model_dump(exclude_unset=True)
|
|
sc = await static_connection_crud.update(db, sc, changes)
|
|
await audit_crud.log(
|
|
db, actor=actor, action="update", resource_type="StaticConnection",
|
|
resource_id=sc.id,
|
|
description=audit_crud.describe_update("StaticConnection", sc.id, changes),
|
|
)
|
|
return sc
|
|
|
|
|
|
@router.delete("/{static_connection_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_static_connection(
|
|
static_connection_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
actor: User = Depends(require_permission("static_connections:delete")),
|
|
):
|
|
sc = await static_connection_crud.get(db, static_connection_id)
|
|
if sc is None:
|
|
raise HTTPException(status_code=404, detail="Static connection not found")
|
|
od_mm = sc.od_mm
|
|
await static_connection_crud.delete(db, sc)
|
|
await audit_crud.log(
|
|
db, actor=actor, action="delete", resource_type="StaticConnection",
|
|
resource_id=static_connection_id,
|
|
description=audit_crud.describe_delete(
|
|
"StaticConnection", static_connection_id, f"{od_mm} mm"
|
|
),
|
|
)
|