-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
45 lines (33 loc) · 1.95 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
from typing import TYPE_CHECKING, List
import fastapi as _fastapi
import sqlalchemy.orm as _orm
import schemas as _schemas
import services as _services
if TYPE_CHECKING:
from sqlalchemy.orm import Session
app = _fastapi.FastAPI()
@app.post("/api/contacts/", response_model= _schemas.Contact)
async def create_contact(contact: _schemas.CreateContact, db: _orm.Session = _fastapi.Depends(_services.get_db), ):
return await _services.create_contact(contact=contact, db=db)
@app.get("/api/contacts/", response_model=List[_schemas.Contact])
async def get_contacts(db:_orm.Session = _fastapi.Depends(_services.get_db)):
return await _services.get_all_contacts(db=db)
@app.get("/api/contacts/{contact_id}/", response_model = _schemas.Contact)
async def get_contact(contact_id: int, db:_orm.Session = _fastapi.Depends(_services.get_db)):
contact = await _services.get_contact(db=db, contact_id=contact_id)
if contact is None:
raise _fastapi.HTTPException(status_code=404, detail="Contact with id does not exist")
return contact
@app.delete("/api/contacts/{contact_id}/")
async def delete_contact(contact_id: int, db:_orm.Session = _fastapi.Depends(_services.get_db)):
contact = await _services.get_contact(db = db, contact_id = contact_id)
if contact is None:
raise _fastapi.HTTPException(status_code=404, detail="Contact with id does not exist")
await _services.delete_contact(contact, db=db)
return "Successfully deleted the user"
@app.put("/api/contacts/{contact_id}/", response_model=_schemas.Contact)
async def update_contact(contact_id: int, contact_data: _schemas.CreateContact, db: _orm.Session = _fastapi.Depends(_services.get_db)):
contact = await _services.get_contact(db=db, contact_id=contact_id)
if contact is None:
raise _fastapi.HTTPException(status_code=404, detail="Contact with id does not exist")
return await _services.update_contact(contact_data=contact_data, contact=contact, db=db)