Hermes Agent
80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
from typing import Any, Type
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import get_current_user
|
|
from app.core.database import get_db
|
|
from app.models import User
|
|
from app.schemas.common import Page, ResponseModel
|
|
|
|
|
|
def crud_router(
|
|
*,
|
|
prefix: str,
|
|
tags: list[str],
|
|
model: Type,
|
|
create_schema: Type[BaseModel],
|
|
update_schema: Type[BaseModel],
|
|
out_schema: Type[BaseModel],
|
|
) -> APIRouter:
|
|
router = APIRouter(prefix=prefix, tags=tags, dependencies=[Depends(get_current_user)])
|
|
|
|
@router.get("", response_model=Page[out_schema]) # type: ignore[valid-type]
|
|
async def list_items(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
keyword: str | None = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
conditions = [model.is_deleted == False]
|
|
if keyword and hasattr(model, "name"):
|
|
conditions.append(model.name.ilike(f"%{keyword}%"))
|
|
query = select(model).where(*conditions)
|
|
total = await db.scalar(select(func.count()).select_from(query.subquery()))
|
|
result = await db.execute(
|
|
query.order_by(model.sort.desc(), model.created_at.desc()).offset((page - 1) * page_size).limit(page_size)
|
|
)
|
|
return Page(items=result.scalars().all(), total=total or 0, page=page, page_size=page_size)
|
|
|
|
@router.get("/{item_id}", response_model=out_schema) # type: ignore[valid-type]
|
|
async def get_item(item_id: str, db: AsyncSession = Depends(get_db)):
|
|
item = await db.get(model, item_id)
|
|
if not item or item.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
return item
|
|
|
|
@router.post("", response_model=out_schema) # type: ignore[valid-type]
|
|
async def create_item(payload: create_schema, db: AsyncSession = Depends(get_db)): # type: ignore[valid-type]
|
|
item = model(**payload.model_dump())
|
|
db.add(item)
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
return item
|
|
|
|
@router.put("/{item_id}", response_model=out_schema) # type: ignore[valid-type]
|
|
async def update_item(item_id: str, payload: update_schema, db: AsyncSession = Depends(get_db)): # type: ignore[valid-type]
|
|
item = await db.get(model, item_id)
|
|
if not item or item.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
data: dict[str, Any] = payload.model_dump(exclude_unset=True)
|
|
for key, value in data.items():
|
|
if hasattr(item, key):
|
|
setattr(item, key, value)
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
return item
|
|
|
|
@router.delete("/{item_id}", response_model=ResponseModel)
|
|
async def delete_item(item_id: str, db: AsyncSession = Depends(get_db)):
|
|
item = await db.get(model, item_id)
|
|
if not item or item.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
item.is_deleted = True
|
|
await db.commit()
|
|
return ResponseModel(message="deleted")
|
|
|
|
return router
|