admin / base_rest.py

from typing import List from fastapi import APIRouter, HTTPException import peewee as pw from peewee import DoesNotExist from pydantic import BaseModel class BaseRest: def __init__(self, model: pw.Model, pydantic_model: BaseModel, methods: List[str]=None): self.model = model self.pydantic_model = pydantic_model self.model_class_name = model._meta.name self.router = APIRouter() self.openapi_extra = { "requestBody": { "content": { "application/json": { "schema": self.pydantic_model.model_json_schema() } }, "required": True, }, } if methods is None: methods = ["list", "create", "get", "update", "delete"] if "list" in methods: self.router.add_api_route( path=f"{self.model_class_name}", endpoint=self.list, response_model=List[self.pydantic_model], methods=["GET"] ) if "create" in methods: self.router.add_api_route( path=f"{self.model_class_name}", endpoint=self.create, openapi_extra=self.openapi_extra, methods=["POST"] ) if "get" in methods: self.router.add_api_route( path=f"{self.model_class_name}/{{{self.model_class_name}_id}}", endpoint=self.get, response_model=self.pydantic_model, methods=["GET"] ) if "update" in methods: self.router.add_api_route( path=f"{self.model_class_name}/{{{self.model_class_name}_id}}", endpoint=self.update, openapi_extra=self.openapi_extra, methods=["PUT"] ) if "delete" in methods: self.router.add_api_route( path=f"{self.model_class_name}/{{{self.model_class_name}_id}}", endpoint=self.delete, methods=["DELETE"] ) # Метод для получения списка объектов async def list(self): try: return list(self.model.select()) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Метод для создания нового объекта async def create(self, data: BaseModel): try: # Преобразуем Pydantic модель в словарь data_dict = data.dict() # Создаем новый объект instance = self.model.create(**data_dict) return instance except Exception as e: raise HTTPException(status_code=400, detail=str(e)) # Метод для получения объекта по ID async def get(self, id: int): try: instance = self.model.get_by_id(id) return instance except DoesNotExist: raise HTTPException(status_code=404, detail="Объект не найден") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Метод для обновления объекта async def update(self, id: int, data: BaseModel): try: # Получаем объект instance = self.model.get_by_id(id) # Обновляем поля for field, value in data.model_dump().items(): setattr(instance, field, value) instance.save() return instance except DoesNotExist: raise HTTPException(status_code=404, detail="Объект не найден") except Exception as e: raise HTTPException(status_code=400, detail=str(e)) # Метод для удаления объекта async def delete(self, id: int): try: # Получаем объект instance = self.model.get_by_id(id) # Удаляем его instance.delete_instance() return {"message": "Объект успешно удален"} except DoesNotExist: raise HTTPException(status_code=404, detail="Объект не найден") except Exception as e: raise HTTPException(status_code=500, detail=str(e))