109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
from typing import Optional
|
|
|
|
from config.module import MODULES, Module
|
|
from core import crud
|
|
from core.database import Base, engine
|
|
from core.util import get_session
|
|
from fastapi import APIRouter, Depends, FastAPI, HTTPException, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
from sqlalchemy.orm import Session
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
def get_router(module: Module) -> APIRouter:
|
|
cls = module.get_item("model")
|
|
Item = module.get_item("schema")
|
|
ItemCreate = module.get_item("create_schema")
|
|
|
|
router = APIRouter(prefix="/" + module.title, tags=[module.title])
|
|
|
|
@router.post("/", response_model=Item)
|
|
async def create_item(item: ItemCreate, db: Session = Depends(get_session)):
|
|
return crud.add_item(db, cls, item)
|
|
|
|
@router.get("/", response_model=list[Item])
|
|
async def list_item(
|
|
skip: int = 0, limit: int = 100, db: Session = Depends(get_session)
|
|
):
|
|
items = crud.view_items(db, cls, skip, limit)
|
|
return items
|
|
|
|
@router.get("/{item_id}", response_model=Item)
|
|
async def read_item(item_id: int, db: Session = Depends(get_session)):
|
|
db_item = crud.view_item(db, cls, item_id)
|
|
if db_item is None:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
return db_item
|
|
|
|
@router.put("/{item_id}", response_model=Item)
|
|
async def update_item(
|
|
item_id: int,
|
|
item: ItemCreate,
|
|
db: Session = Depends(get_session),
|
|
):
|
|
return crud.edit_item(db, cls, item_id, item)
|
|
|
|
@router.delete("/{item_id}", response_model=Item)
|
|
async def delete_item(item_id: int, db: Session = Depends(get_session)):
|
|
db_item = crud.view_item(db, cls, item_id)
|
|
if db_item is None:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
for field in db_item._file_fields:
|
|
crud.remove_file(db, cls, item_id, field)
|
|
crud.remove_item(db, cls, item_id)
|
|
|
|
return db_item
|
|
|
|
@router.get("/{item_id}/{field}", response_class=FileResponse)
|
|
async def read_file(item_id: int, field: str, db: Session = Depends(get_session)):
|
|
db_item = crud.view_item(db, cls, item_id)
|
|
return getattr(db_item, field)
|
|
|
|
@router.post("/{item_id}/{field}", response_model=Optional[str])
|
|
async def create_file(
|
|
item_id: int, field: str, file: UploadFile, db: Session = Depends(get_session)
|
|
):
|
|
try:
|
|
db_item = crud.add_file(db, cls, item_id, field, file)
|
|
|
|
except Exception:
|
|
raise HTTPException(status_code=500, detail="Can not upload file")
|
|
|
|
finally:
|
|
file.file.close()
|
|
|
|
return getattr(db_item, field)
|
|
|
|
@router.delete("/{item_id}/{field}", response_model=Item)
|
|
async def delete_file(item_id: int, field: str, db: Session = Depends(get_session)):
|
|
db_item = crud.remove_file(db, cls, item_id, field)
|
|
return db_item
|
|
|
|
return router
|
|
|
|
|
|
for module in tuple(Module(module_name) for module_name in MODULES):
|
|
app.include_router(get_router(module))
|
|
|
|
module_router = APIRouter(prefix="/_module", tags=["_module"])
|
|
|
|
|
|
@module_router.get("/", response_model=list[str])
|
|
def list_module(skip: int = 0, limit: int = 100):
|
|
return MODULES
|
|
|
|
|
|
@module_router.get("/{module}", response_model=dict)
|
|
def read_module(module: str):
|
|
if module not in MODULES:
|
|
raise HTTPException(status_code=404, detail="Module not found")
|
|
return Module(module).get_body()
|
|
|
|
|
|
app.include_router(module_router)
|
|
|
|
|
|
Base.metadata.create_all(bind=engine)
|