smolkik-code
9b15f8b09c
- PostgreSQL schema: 14 tables, JSONB attributes, audit triggers, soft delete - FastAPI backend: CRUD, search/filter, relationship graph, bulk import, JWT RBAC - React frontend: CI table, detail card, force-graph, dashboard - Seed data: homelab scenario (Proxmox, Mikrotik, VMs, services) - Docker Compose + Kubernetes manifests - 20 backend tests (pytest + httpx)
61 lines
2.6 KiB
Python
61 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from app.database import get_db
|
|
from app.models.models import CIClass, CIType, Location, User
|
|
from app.schemas.schemas import (
|
|
CIClassCreate, CIClassResponse, CITypeCreate, CITypeResponse,
|
|
LocationCreate, LocationResponse
|
|
)
|
|
from app.middleware.auth import require_viewer, require_editor
|
|
|
|
router = APIRouter(prefix="/api", tags=["reference"])
|
|
|
|
|
|
@router.get("/classes", response_model=list[CIClassResponse])
|
|
async def list_classes(db: AsyncSession = Depends(get_db), _user: User = Depends(require_viewer)):
|
|
result = await db.execute(select(CIClass).where(CIClass.deleted_at.is_(None)).order_by(CIClass.name))
|
|
return [CIClassResponse.model_validate(c) for c in result.scalars().all()]
|
|
|
|
|
|
@router.post("/classes", response_model=CIClassResponse, status_code=201)
|
|
async def create_class(body: CIClassCreate, db: AsyncSession = Depends(get_db), _user: User = Depends(require_editor)):
|
|
cls = CIClass(**body.model_dump())
|
|
db.add(cls)
|
|
await db.flush()
|
|
await db.refresh(cls)
|
|
return CIClassResponse.model_validate(cls)
|
|
|
|
|
|
@router.get("/types", response_model=list[CITypeResponse])
|
|
async def list_types(class_id: str = None, db: AsyncSession = Depends(get_db), _user: User = Depends(require_viewer)):
|
|
q = select(CIType).where(CIType.deleted_at.is_(None))
|
|
if class_id:
|
|
q = q.where(CIType.class_id == class_id)
|
|
result = await db.execute(q.order_by(CIType.name))
|
|
return [CITypeResponse.model_validate(t) for t in result.scalars().all()]
|
|
|
|
|
|
@router.post("/types", response_model=CITypeResponse, status_code=201)
|
|
async def create_type(body: CITypeCreate, db: AsyncSession = Depends(get_db), _user: User = Depends(require_editor)):
|
|
ct = CIType(**body.model_dump())
|
|
db.add(ct)
|
|
await db.flush()
|
|
await db.refresh(ct)
|
|
return CITypeResponse.model_validate(ct)
|
|
|
|
|
|
@router.get("/locations", response_model=list[LocationResponse])
|
|
async def list_locations(db: AsyncSession = Depends(get_db), _user: User = Depends(require_viewer)):
|
|
result = await db.execute(select(Location).where(Location.deleted_at.is_(None)).order_by(Location.name))
|
|
return [LocationResponse.model_validate(l) for l in result.scalars().all()]
|
|
|
|
|
|
@router.post("/locations", response_model=LocationResponse, status_code=201)
|
|
async def create_location(body: LocationCreate, db: AsyncSession = Depends(get_db), _user: User = Depends(require_editor)):
|
|
loc = Location(**body.model_dump())
|
|
db.add(loc)
|
|
await db.flush()
|
|
await db.refresh(loc)
|
|
return LocationResponse.model_validate(loc)
|