/
main.py
188 lines (131 loc) · 5.1 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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
from typing import Type, Annotated
import bcrypt
import uvicorn
import jwt
import datetime
from dotenv import dotenv_values
from fastapi import FastAPI, Depends, Header
from sqlalchemy.orm import Session
from sqlalchemy.sql import func, desc
from database.connection import connect, engine
from models.question import CreateQuestion, QuestionType
from models.quiz import GetQuiz, CreateQuiz
from schemas.quiz import Quiz
from models.auth import Login, SignUp
from schemas.question import Question
from schemas.answer import Answer
from schemas.user import User
from schemas.base import Base
from fastapi.middleware.cors import CORSMiddleware
env = dotenv_values(".env")
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
def generate_token(data: int):
token = jwt.encode(
{"user": data, 'typ': 'JWT', "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)},
env.get("JWT_SECRET"), algorithm="HS256",
)
return token
@app.post("/create_question")
async def create_question(data: CreateQuestion, database: Session = Depends(connect)):
new_question = Question(title=data.title, tip=data.tip, points=data.points, type=data.type, thumb=data.thumb)
database.add(new_question)
database.commit()
for answer in data.answers:
new_answer = Answer(title=answer.title, correct=answer.correct, question_id=new_question.id)
database.add(new_answer)
database.commit()
question = database.query(Question).where(Question.id == new_question.id).first()
return {**question.__dict__, "answers": question.answers}
@app.post("/login")
async def login(data: Login, database: Session = Depends(connect)):
try:
user: Type[User] = database.query(User).where(User.username == data.username).first()
if user is None:
raise ValueError
match = bcrypt.checkpw(data.password.encode('utf-8'), user.password)
if match:
token = generate_token(user.id)
user.__dict__.pop("password")
return {"user": user.__dict__, "token": token}
else:
raise ValueError
except ValueError:
return {"error": True, "message": "Invalid credentials"}
@app.post("/signup")
async def signup(data: SignUp, database: Session = Depends(connect)):
try:
exits = database.query(User).where(User.username == data.username).first()
if exits is None:
password = bcrypt.hashpw(data.password.encode('utf-8'), bcrypt.gensalt())
user = User(username=data.username, avatar=data.avatar, password=password)
database.add(user)
database.commit()
token = generate_token(user.id)
user.__dict__.pop("password")
return {"user": user.__dict__, "token": token}
raise ValueError
except ValueError:
return {"error": True, "message": "User already exist"}
@app.get("/auth")
async def auth(authorization: Annotated[str | None, Header()] = None, database: Session = Depends(connect)):
try:
data = jwt.decode(
str(authorization).encode('utf-8'),
key=env.get("JWT_SECRET"),
algorithms=["HS256"],
)
user = database.query(User).where(User.id == data['user']).first()
user.__dict__.pop("password")
return user.__dict__
except Exception as e:
print(e)
return {"error": True, "message": "Session expired"}
@app.post("/get_quiz")
async def get_quiz(data: GetQuiz, database: Session = Depends(connect)):
global questions
if data.type == QuestionType.NONE:
questions = (database.query(Question))
else:
questions = (database.query(Question).filter(Question.type == data.type))
res = []
for question in questions.order_by(func.random()).limit(5).all():
res.append({**question.__dict__, "answers": question.answers})
return res
@app.post("/create_quiz")
async def create_quiz(data: CreateQuiz, database: Session = Depends(connect)):
exist = (database.query(Quiz).where(Quiz.user_id == data.user_id).first())
if exist is not None:
exist.points = data.points
exist.time = data.time
exist.type = data.type
database.commit()
return exist
else:
new_quiz = Quiz(user_id=data.user_id, time=data.time, points=data.points, type=data.type)
database.add(new_quiz)
database.commit()
return new_quiz
@app.get("/ranking")
async def ranking(database: Session = Depends(connect)):
history = database.query(Quiz).order_by(desc(Quiz.points)).limit(70)
data = history.all()
res = []
for i in data:
user = database.query(User).where(User.id == i.user_id).first()
res.append({**i.__dict__, "username": user.username, "avatar": user.avatar})
return res
@app.get("/healthcheck")
def healthcheck():
return {"ok": True}
@app.on_event("startup")
def startup():
Base.metadata.create_all(bind=engine)
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=9696)