surtur
fc0051b243
* aside from slight refactors there have been route changes (/api/v1/service/ --> /api/v1/service) * added helper little function for grabbing user and service ids * refactored a couple of return statements * added separate models for updates * adjusted error handling behaviour: more general 422 is now returned (previously 404 was used) --> hopefully leaking less information * updates are divided into multiple queries **as of now**, mainly due to my sqlalchemy amateurism: can't figure out how to construct a proper multi-column-update query programatically
117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
# vim:fileencoding=utf-8:ft=python:foldmethod=marker
|
|
# statuspage/app/main.py
|
|
|
|
from typing import List, Optional
|
|
|
|
from fastapi import FastAPI, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import crud, models, schemas
|
|
from app.database import SessionLocal, engine
|
|
|
|
|
|
models.Base.metadata.create_all(bind=engine)
|
|
|
|
app = FastAPI()
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
|
|
|
|
def get_usr(user_id: int, db: Session):
|
|
db=db
|
|
user_id=user_id
|
|
db_user = crud.get_user(db=db, user_id=user_id)
|
|
return db_user
|
|
|
|
def get_srv(service_id: int, db: Session):
|
|
db=db
|
|
service_id=service_id
|
|
db_service = crud.get_service(db=db, service_id=service_id)
|
|
return db_service
|
|
|
|
|
|
|
|
@app.get("/api/v1/ping")
|
|
def pong():
|
|
return {"ping": "pong!"}
|
|
|
|
@app.get("/api/v1/service")
|
|
async def read_services(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
return crud.get_services(db, skip=skip, limit=limit)
|
|
|
|
@app.post("/api/v1/service", response_model=schemas.Service)
|
|
async def create_service(service: schemas.ServiceCreate, db: Session = Depends(get_db)):
|
|
return crud.create_service(db=db, service=service)
|
|
|
|
@app.get("/api/v1/service/{service_id}", response_model=schemas.Service)
|
|
def read_user(service_id: int, db: Session = Depends(get_db)):
|
|
db_service = get_srv(service_id, db)
|
|
if db_service is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return db_service
|
|
|
|
@app.put("/api/v1/service/{service_id}")
|
|
async def update_user(service_id: int, service: schemas.ServiceUpdate, db: Session = Depends(get_db)):
|
|
if ((service.name is None) and (service.owner_id is None) and (service.is_private is None) and (service.description is None) and (service.service_type is None) and (service.url is None) and (service.is_active is None)):
|
|
raise HTTPException(status_code=400, detail="No data provided")
|
|
db_service = get_srv(service_id, db)
|
|
if db_service is None:
|
|
raise HTTPException(status_code=422, detail="Unprocessable entity")
|
|
crud.update_service(db=db, service_id=service_id, s=service)
|
|
return {"update":"success"}
|
|
|
|
@app.delete("/api/v1/service/{service_id}")
|
|
async def del_user(service_id: int, db: Session = Depends(get_db)):
|
|
db_service = get_srv(service_id, db)
|
|
if db_service is None:
|
|
raise HTTPException(status_code=422, detail="Unprocessable entity")
|
|
crud.del_service(db=db, service_id=service_id)
|
|
return {"delete":"success"}
|
|
|
|
|
|
|
|
@app.post("/api/v1/users", response_model=schemas.User)
|
|
async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
|
db_user = crud.get_user_by_username(db, name=user.name)
|
|
if db_user:
|
|
raise HTTPException(status_code=400, detail="username already registered")
|
|
return crud.create_user(db=db, user=user)
|
|
|
|
@app.get("/api/v1/users", response_model=List[schemas.User])
|
|
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
return crud.get_users(db=db, skip=skip, limit=limit)
|
|
|
|
@app.get("/api/v1/users/{user_id}", response_model=schemas.User)
|
|
def read_user(user_id: int, db: Session = Depends(get_db)):
|
|
db_user = get_usr(user_id, db)
|
|
if db_user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return db_user
|
|
|
|
@app.put("/api/v1/users/{user_id}")
|
|
async def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
|
|
if ((user.name is None) and (user.password is None) and (user.full_name is None) and (user.is_active is None)):
|
|
raise HTTPException(status_code=400, detail="No data provided")
|
|
db_user = read_usr(user_id, db)
|
|
if db_user is None:
|
|
raise HTTPException(status_code=422, detail="Unprocessable entity")
|
|
crud.update_user(db=db, user_id=user_id, user=user)
|
|
return {"update":"success"}
|
|
|
|
@app.delete("/api/v1/users/{user_id}")
|
|
async def del_user(user_id: int, db: Session = Depends(get_db)):
|
|
db_user = read_usr(user_id, db)
|
|
if db_user is None:
|
|
raise HTTPException(status_code=422, detail="Unprocessable entity")
|
|
crud.del_user(db=db, user_id=user_id)
|
|
return {"delete":"success"}
|
|
|