This repository has been archived on 2020-08-14. You can view files and clone it, but cannot push or open issues or pull requests.
statuspage/app/main.py
surtur 2dd2258472
feat: {user,service} {creation,deletion,listing}
* also updated requirements.txt accordingly
2020-08-04 23:27:13 +02:00

68 lines
2.0 KiB
Python

# vim:fileencoding=utf-8:ft=python:foldmethod=marker
# statuspage/app/main.py
from typing import List
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@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.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])
@app.get("/api/v1/users")
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 = crud.get_user(db=db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
@app.delete("/api/v1/users/{user_id}")
async def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db=db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
crud.del_user(db=db, user_id=user_id)
return {"deletion":"successful"}