-
Notifications
You must be signed in to change notification settings - Fork 7
/
main.py
113 lines (81 loc) · 2.3 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
import datetime
from fastapi import FastAPI, Depends
from pydantic import BaseModel
import databases
from sqlalchemy import engine_from_config, pool
from sqlalchemy.orm import Session
from logging.config import fileConfig
from alembic import context
from typing import List
from db.models import models
from db.db import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()
"""
username
email
created_at
updated_at
"""
class User(BaseModel):
username: str
email: str
created_at: datetime.datetime
updated_at: datetime.datetime
class UserCreate(BaseModel):
username: str
email: str
"""
title
type: ChoiceField (e.g. text or image)
created_by
created_at
updated_at
is_add_choices_active
is_voting_active
"""
class Poll(BaseModel):
title: str
type: str
is_add_choices_active: bool
is_voting_active: bool
created_by: int
created_at: datetime.datetime
updated_at: datetime.datetime
def create_user(db: Session, user: UserCreate):
db_user = models.User(email=user.email, username=user.username)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/polls")
async def root():
return {"polls": "Hello World"}
@app.get("/users/", response_model=List[User])
def get_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = get_users(db, skip=skip, limit=limit)
return users
@app.post("/users/", response_model=User)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return create_user(db=db, user=user)
@app.post("/polls/")
async def create_poll(poll: Poll):
return poll