-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
124 lines (84 loc) · 3.58 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
from database import SessionLocal, engine
import schemas
import models
import crud
from sqlalchemy.orm import Session
from fastapi import Depends, FastAPI, HTTPException
from fastapi.openapi.utils import get_openapi
import re
PATTERN = re.compile("^([a-z]+[,.]?[ ]?|[a-z]+['-]?)+$")
models.Base.metadata.create_all(bind=engine)
app = FastAPI(openapi_url="/api/v1/openapi.json",
docs_url="/api/livedocumentation", redoc_url=None)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
PATTERN_ERROR_MSG = "Invalid name input"
NOT_FOUND_ERROR_MEG = "Person not found"
ALREADY_EXIST_MSG = "Name exist already"
@app.post("/api", response_model=schemas.Person, tags=["persons"])
def create_person(person: schemas.PersonBase, db: Session = Depends(get_db)):
if not PATTERN.match(person.name):
raise HTTPException(
status_code=400, detail=PATTERN_ERROR_MSG)
new_person = crud.get_person_by_name(db, name=person.name)
if new_person:
raise HTTPException(status_code=400, detail=ALREADY_EXIST_MSG)
return crud.create_person(db=db, person=person)
@app.get("/", response_model=list[schemas.Person], tags=["persons"])
def index(db: Session = Depends(get_db)):
persons = crud.get_persons(db)
return persons
@app.get("/api/{user_id}", response_model=schemas.Person, tags=["persons"])
def read_person(user_id: int, db: Session = Depends(get_db)):
db_person = crud.get_person(db, user_id=user_id)
if db_person is None:
raise HTTPException(status_code=404, detail=NOT_FOUND_ERROR_MEG)
return db_person
@app.get("/api", response_model=schemas.Person, tags=["persons"])
def read_person_by_name(name: str, db: Session = Depends(get_db)):
db_person = crud.get_person_by_name(db, name=name)
if db_person is None:
raise HTTPException(status_code=404, detail=NOT_FOUND_ERROR_MEG)
return db_person
@app.put("/api/{user_id}", response_model=schemas.Person, tags=["persons"])
def update_person(user_id: int, edit_person: schemas.PersonBase, db: Session = Depends(get_db)):
if not PATTERN.match(edit_person.name):
raise HTTPException(
status_code=400, detail=PATTERN_ERROR_MSG)
db_person = crud.get_person(db, user_id=user_id)
if db_person is None:
raise HTTPException(status_code=404, detail=NOT_FOUND_ERROR_MEG)
if db_person.name == edit_person.name.title():
raise HTTPException(
status_code=400, detail="No change in the name")
get_edit_name = crud.get_person_by_name(db, name=edit_person.name)
if get_edit_name:
raise HTTPException(status_code=400, detail=ALREADY_EXIST_MSG)
return crud.edit_person(db=db, user_id=user_id, edit_person=edit_person)
@app.delete("/api/{user_id}", tags=["persons"])
def delete_person(user_id: int, db: Session = Depends(get_db)):
db_person = crud.get_person(db, user_id=user_id)
if db_person is None:
raise HTTPException(
status_code=404, detail=f"Person with id number: {user_id}, not found.")
return crud.delete_person(db, user_id=user_id)
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Person API",
version="1.0.0",
summary="This is a RESTful API with complete CRUD functionality",
routes=app.routes,
)
openapi_schema["info"]["x-logo"] = {
"url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi