Newer
Older
safe-algo-pro / apis / model.py
zhangyingjie on 12 Oct 2 KB first commit
from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlmodel import Session

from apis.base import standard_response, StandardResponse, PageResponse, standard_error_response
from common.biz_exception import BizException
from db.database import get_db
from entity.model import AlgoModel, AlgoModelCreate, AlgoModelUpdate, AlgoModelInfo
from services.model_service import ModelService

router = APIRouter()


@router.get("/list", response_model=StandardResponse[List[AlgoModelInfo]])
def get_model_list(
        name: Optional[str] = None,
        remark: Optional[str] = None,
        db: Session = Depends(get_db)):
    service = ModelService(db)
    models = list(service.get_model_list(name, remark))
    return standard_response(data=models)


@router.get("/page/", response_model=StandardResponse[PageResponse[AlgoModel]])
def get_model_page(
        name: Optional[str] = None,
        remark: Optional[str] = None,
        offset: int = Query(0, ge=0),  # 从第几条开始
        limit: int = Query(10, ge=1),  # 每页显示多少条记录
        db: Session = Depends(get_db)):
    service = ModelService(db)

    # 获取分页后的设备列表和总数
    models, total = service.get_model_page(name, remark, offset, limit)

    return standard_response(
        data=PageResponse(total=total, items=models)
    )


@router.post("/add", response_model=StandardResponse[AlgoModelInfo])
def create_model(model_data: AlgoModelCreate, db: Session = Depends(get_db)):
    service = ModelService(db)
    model = service.create_model(model_data)
    return standard_response(data=model)


@router.post("/update", response_model=StandardResponse[AlgoModelInfo])
def update_model(model_data: AlgoModelUpdate, db: Session = Depends(get_db)):
    service = ModelService(db)
    model = service.update_model(model_data)
    if not model:
        return standard_error_response(data=model_data, message="Model not found")
    return standard_response(data=model)


@router.delete("/delete", response_model=StandardResponse[int])
def delete_model(model_id: int, db: Session = Depends(get_db)):
    service = ModelService(db)
    model = service.delete_model(model_id)
    if not model:
        return standard_error_response(data=model_id, message="Model not found")
    return standard_response(data=model_id)