statuspage/app/main.py

# vim:fileencoding=utf-8:ft=python:foldmethod=marker
# statuspage/app/main.py

from typing import List, Optional

from fastapi import FastAPI, Depends, HTTPException
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")
async def pong():
	"""Basic sanity check"""
	return {"ping": "pong!"}

@app.get("/api/v1/service")
async def read_services(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
	"""List all services"""
	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)):
	"""Service types: icmp = 0, http = 1"""
	if service.service_type not in [0, 1]:
		raise HTTPException(status_code=422, detail="Invalid service type provided")
	owner = get_usr(service.owner_id, db)
	if owner is None:
		raise HTTPException(status_code=422, detail="No such owner_id")
	return crud.create_service(db=db, service=service)

@app.get("/api/v1/service/{service_id}", response_model=schemas.Service)
async def read_service(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="Service not found")
	return db_service

@app.patch("/api/v1/service/{service_id}")
async def update_service(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 delete_service(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)):
	"""List all users"""
	return crud.get_users(db=db, skip=skip, limit=limit)

@app.get("/api/v1/users/{user_id}", response_model=schemas.User)
async 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.patch("/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 = get_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 delete_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=422, detail="Unprocessable entity")
	crud.del_user(db=db, user_id=user_id)
	return {"delete":"success"}