LobotomyLabs/classquiz-ai
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Mawoka
2022-02-27 21:23:57 +01:00
commit 1805d6fe98
46 changed files with 4453 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
classquiz/__init__.py
+28
View File
@@ -0,0 +1,28 @@
from fastapi import FastAPI
from classquiz.socket_server import sio
from socketio import ASGIApp
from classquiz.routers import users, quiz
from classquiz.db import database
app = FastAPI()
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
app.include_router(users.router, tags=["users"], prefix="/api/v1/users")
app.include_router(quiz.router, tags=["quiz"], prefix="/api/v1/quiz")
app.mount("/", ASGIApp(sio))
classquiz/auth.py
+147
View File
@@ -0,0 +1,147 @@
from classquiz.config import settings
from classquiz.db.models import *
from typing import Union
from jose import JWTError, jwt
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from datetime import datetime, timedelta
from classquiz.cache import get_cache
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2
from fastapi.openapi.models import OAuthFlows as OAuthFlowsModel
from fastapi import Request
from fastapi.security.utils import get_authorization_scheme_param
from fastapi import HTTPException
from fastapi import status
from typing import Optional
from typing import Dict
from starlette.status import HTTP_401_UNAUTHORIZED
import pydantic
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = settings.access_token_expire_minutes
class OAuth2PasswordBearerWithCookie(OAuth2):
def __init__(
self,
tokenUrl: str,
scheme_name: Optional[str] = None,
scopes: Optional[Dict[str, str]] = None,
auto_error: bool = True,
):
if not scopes:
scopes = {}
flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes})
super().__init__(flows=flows, scheme_name=scheme_name, auto_error=auto_error)
async def __call__(self, request: Request) -> Optional[str]:
authorization: str = request.cookies.get("access_token") # changed to accept access token from httpOnly Cookie
scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
if self.auto_error:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
else:
return None
return param
oauth2_scheme = OAuth2PasswordBearerWithCookie(tokenUrl="/api/v1/users/token/cookie")
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
async def get_user_from_mail(email: str) -> Union[User, None]:
return await get_cache(criteria="email", content=email)
async def get_user_from_username(username: str) -> Union[User, None]:
return await get_cache(criteria="username", content=username)
async def get_user_from_id(id: str) -> Union[User, None]:
return await get_cache(criteria="id", content=id)
async def authenticate_user(email: str, password: str) -> Union[User, bool]:
user = await get_user_from_mail(email)
if not user:
return False
if not verify_password(password, user.password):
return False
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise credentials_exception
token_data = TokenData(email=email)
except JWTError:
raise credentials_exception
user = await get_user_from_mail(email=token_data.email)
if user is None:
raise credentials_exception
return user
async def get_current_user_optional(token: str = Depends(oauth2_scheme)) -> User | None:
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
return None
token_data = TokenData(email=email)
except JWTError:
return None
user = await get_user_from_mail(email=token_data.email)
if user is None:
return None
return user
async def check_token(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise credentials_exception
token_data = TokenData(email=email)
except JWTError:
raise credentials_exception
return token_data.email
classquiz/cache.py
+52
View File
@@ -0,0 +1,52 @@
from classquiz.config import redis, settings
from classquiz.db.models import User
import ormar
import uuid
from typing import Union
from json import loads, dumps
async def cache_account(criteria: str, content: str) -> Union[User, None]:
async def insert_into_redis(usermodel: User, key: str):
await redis.set(key, usermodel.json(), ex=settings.cache_expiry)
if criteria == "email":
try:
res = await User.objects.get(email=content)
except ormar.exceptions.NoMatch:
return None
await insert_into_redis(res, content)
return res
elif criteria == "username":
try:
res = await User.objects.get(username=content)
except ormar.exceptions.NoMatch:
return None
await insert_into_redis(res, content)
return res
elif criteria == "id":
try:
res = await User.objects.get(id=uuid.UUID(content))
except ormar.exceptions.NoMatch:
return None
await insert_into_redis(res, content)
return res
else:
return None
async def get_from_redis(key: str) -> Union[None, User]:
user = await redis.get(key)
if user is None:
return None
else:
return User.parse_obj(loads(user))
async def get_cache(criteria: str, content: str) -> User:
cache = await get_from_redis(content)
if cache is not None:
return cache
else:
return await cache_account(criteria, content)
classquiz/config.py
+38
View File
@@ -0,0 +1,38 @@
from pydantic import BaseSettings, RedisDsn
import redis.asyncio as redis_lib
class Settings(BaseSettings):
"""
Settings class for the shop app.
"""
root_address: str = "http://127.0.0.1:8000"
redis: RedisDsn = "redis://localhost:6379/0?decode_responses=True"
skip_email_verification: bool = False
db_url: str = "sqlite:///classquiz.db"
mail_address: str
mail_password: str
mail_username: str
mail_server: str
mail_port: int
secret_key: str
minio_url: str = "127.0.0.1"
minio_access_key: str
minio_secret_key: str
minio_bucket: str
minio_secure: bool = True
access_token_expire_minutes: int = 30
cache_expiry: int = 86400
typesense_api_key: str
typesense_host: str = "localhost"
typesense_port: int = 8108
typesense_protocol: str = "http"
typesense_timeout: int = 2
class Config:
env_file = ".env"
env_file_encoding = 'utf-8'
settings = Settings()
redis: redis_lib.client.Redis = redis_lib.Redis().from_url(settings.redis)
classquiz/db/__init__.py
+6
View File
@@ -0,0 +1,6 @@
import databases
import sqlalchemy
from classquiz.config import settings
database = databases.Database(settings.db_url)
metadata = sqlalchemy.MetaData()
classquiz/db/models.py
+115
View File
@@ -0,0 +1,115 @@
import uuid
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Json
from . import metadata, database
import ormar
class User(ormar.Model):
"""
The user model
"""
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4())
email: str = ormar.String(unique=True, max_length=100)
username: str = ormar.String(unique=True, max_length=100)
password: str = ormar.String(unique=True, max_length=100)
verified: bool = ormar.Boolean(default=False)
verify_key: str = ormar.String(unique=True, max_length=100, nullable=True)
created_at: datetime = ormar.DateTime(default=datetime.now())
class Meta:
tablename = 'users'
metadata = metadata
database = database
class UserSession(ormar.Model):
"""
The user session model
"""
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4())
user: uuid.UUID = ormar.ForeignKey(User)
session_key: str = ormar.String(unique=True, max_length=64)
created_at: datetime = ormar.DateTime(default=datetime.now())
class Meta:
tablename = 'user_sessions'
metadata = metadata
database = database
class QuizAnswer(BaseModel):
right: bool
answer: str
class QuizQuestion(BaseModel):
question: str
time: str # in Secs
answers: list[QuizAnswer]
class QuizInput(BaseModel):
public: bool = False
title: str
description: str
questions: list[QuizQuestion]
class Quiz(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4())
public: bool = ormar.Boolean(default=False)
title: str = ormar.String(max_length=100)
description: str = ormar.String(max_length=300, nullable=True)
created_at: datetime = ormar.DateTime(default=datetime.now())
updated_at: datetime = ormar.DateTime(default=datetime.now())
user_id: uuid.UUID = ormar.UUID(foreign_key=User.id)
questions: Json[list[QuizQuestion]] = ormar.JSON(nullable=False)
class Meta:
tablename = 'quiz'
metadata = metadata
database = database
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
email: str | None = None
class PlayGame(BaseModel):
quiz_id: uuid.UUID | str
description: str
title: str
questions: list[QuizQuestion]
game_id: uuid.UUID
game_pin: str
started: bool = False
class GamePlayer(BaseModel):
username: str
sid: str
class GameAnser2(BaseModel):
username: str
right: bool
answer: str
class GameAnser1(BaseModel):
id: int
answers: list[GameAnser2]
class GameSession(BaseModel):
admin: str
game_id: str
players: list[GamePlayer | None]
answers: list[GameAnser1 | None]
classquiz/emails/__init__.py
+4
View File
@@ -0,0 +1,4 @@
async def send_mail(email: str):
pass
classquiz/routers/__init__.py
View File
classquiz/routers/quiz.py
+67
View File
@@ -0,0 +1,67 @@
import json
import os
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from classquiz.config import redis
import pydantic
from classquiz.auth import get_current_user, get_current_user_optional
from classquiz.db.models import Quiz, QuizInput, User, PlayGame
from random import randint
import uuid
router = APIRouter()
@router.post("/create")
async def create_quiz_lol(quiz_input: QuizInput, user: User = Depends(get_current_user)):
quiz = Quiz(**quiz_input.dict(), user_id=user.id)
return await quiz.save()
@router.get("/get/{quiz_id}")
async def get_quiz_from_id(quiz_id: str, user: User | None = Depends(get_current_user_optional)):
try:
quiz_id = uuid.UUID(quiz_id)
except ValueError:
raise HTTPException(status_code=400, detail="badly formed quiz id")
if user is None:
quiz = await Quiz.objects.get_or_none(id=quiz_id, public=True)
else:
quiz = await Quiz.objects.get_or_none(id=quiz_id, public=False, user_id=user.id)
if quiz is None:
return JSONResponse(status_code=404, content={"detail": "quiz not found"})
else:
return quiz
@router.post("/start/{quiz_id}")
async def start_quiz(quiz_id: str, user: User = Depends(get_current_user)):
try:
quiz_id = uuid.UUID(quiz_id)
except ValueError:
raise HTTPException(status_code=400, detail="badly formed quiz id")
quiz = await Quiz.objects.get_or_none(id=quiz_id, public=False, user_id=user.id)
if quiz is None:
return JSONResponse(status_code=404, content={"detail": "quiz not found"})
else:
game_pin = randint(10000000, 99999999)
game = PlayGame(quiz_id=quiz_id, game_pin=str(game_pin), questions=quiz.questions, game_id=uuid.uuid4(),
title=quiz.title, description=quiz.description)
await redis.set(f"game:{str(game.game_pin)}", (game.json()))
return {**quiz.dict(exclude={"id"}), **game.dict(exclude={"questions"})}
@router.get("/join/{game_pin}")
async def get_game_id(game_pin: str):
redis_res = (await redis.get(f"game:{game_pin}")).decode()
if redis_res is None:
raise HTTPException(status_code=404, detail="game not found")
else:
return json.loads(redis_res)["game_id"]
@router.get("/list")
async def get_quiz_list(user: User = Depends(get_current_user)):
return await Quiz.objects.filter(user_id=user.id).all()
classquiz/routers/users.py
+90
View File
@@ -0,0 +1,90 @@
import os
from fastapi import APIRouter, HTTPException, Response
from email_validator import validate_email, EmailNotValidError
from classquiz.auth import *
from fastapi.background import BackgroundTasks
from classquiz.emails import send_mail
from fastapi.responses import JSONResponse
from classquiz.db.models import *
router = APIRouter()
route_user = User.get_pydantic(
exclude={"id": ..., "verified": ..., "verify_key": ..., "created_at": ..., "usersessions": ...})
@router.post("/create", response_model=User,
response_model_include={"id": ..., "verified": ..., "email": ...})
async def create_user(user: route_user, background_task: BackgroundTasks) -> User | JSONResponse:
user = User(**user.dict())
try:
validate_email(user.email)
except EmailNotValidError as e:
raise HTTPException(status_code=400, detail=str(e))
user.verify_key = str(os.urandom(16).hex())
res = await User.objects.filter((User.email == user.email) | (User.username == user.username)).all()
if len(res) != 0:
raise HTTPException(status_code=400, detail="User already exists")
user.password = get_password_hash(user.password)
if len(user.username) == 32:
return JSONResponse({"details": "Username mustn't be 32 characters long"}, 400)
res = await user.save()
background_task.add_task(send_mail, email=user.email)
return user
@router.post("/token/cookie", response_model=Token)
async def login_for_cookie_access_token(request: Request, response: Response,
form_data: OAuth2PasswordRequestForm = Depends()):
user = await authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
session_key = os.urandom(32).hex()
# user_session = UserSession(user=user, session_key=session_key)
# print(user_session, "HALLO!!!!")
# await user_session.save()
access_token_expires = timedelta(minutes=settings.access_token_expire_minutes)
access_token = create_access_token(
data={"sub": user.email}, expires_delta=access_token_expires
)
rememberme_token = create_access_token(
data={"sub": user.email}, expires_delta=timedelta(days=360))
response.set_cookie(key="access_token", value=f"Bearer {access_token}",
httponly=True, samesite='strict')
response.set_cookie(key="expiry", value="", max_age=settings.access_token_expire_minutes * 60)
response.set_cookie(key="rememberme", value="")
response.set_cookie(key="rememberme_token", value=rememberme_token, httponly=True, samesite='strict')
return {"access_token": access_token, "token_type": "bearer"}
@router.get("/token/rememberme")
async def rememberme_token(request: Request, response: Response):
rememberme_cookie = request.cookies.get("rememberme_token")
if rememberme_cookie is None:
raise HTTPException(status_code=400, detail="No rememberme cookie")
payload = jwt.decode(rememberme_cookie, settings.secret_key, algorithms=[ALGORITHM])
access_token_expires = timedelta(minutes=settings.access_token_expire_minutes)
access_token = create_access_token(
data={"sub": payload}, expires_delta=access_token_expires
)
response.set_cookie(key="access_token", value=f"Bearer {access_token}",
httponly=True, samesite='strict')
response.set_cookie(key="expiry", value="", max_age=settings.access_token_expire_minutes * 60)
@router.get("/logout")
async def logout(request: Request, response: Response):
remember_token = request.cookies.get("rememberme_token")
if remember_token is not None:
UserSession.objects.filter(session_key=remember_token).delete()
response.delete_cookie("access_token")
response.delete_cookie("expiry")
response.delete_cookie("rememberme")
response.delete_cookie("rememberme_token")
classquiz/socket_server/__init__.py
+147
View File
@@ -0,0 +1,147 @@
import socketio
import json
from classquiz.config import redis
from classquiz.db.models import GameSession, GameAnser1, GameAnser2, PlayGame
from redis.commands.json.path import Path
sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
@sio.event
async def connect(sid, environ, lol):
print(sid, "connected")
# sio.enter_room(sid, lol)
# print(lol)
# print(environ["asgi.scope"]["headers"])
# print(environ)
# async with sio.session(sid) as session:
# session['username'] = "username"
@sio.event
async def join_game(sid, data):
print(sid, data, "JOIN_GAME")
redis_res = await redis.get(f"game:{data['game_pin']}")
if redis_res is None:
await sio.emit("game_not_found", room=sid)
else:
session = {'game_pin': data["game_pin"], "username": data["username"], "admin": False}
await sio.save_session(sid, session)
await sio.emit("joined_game", redis_res, room=sid)
redis_res = (await redis.get(f"game_session:{data['game_pin']}"))
redis_res = json.loads(redis_res)
print(redis_res)
redis_res["players"].append({"username": data["username"], "sid": sid})
await redis.set(f"game_session:{data['game_pin']}",
json.dumps({"admin": redis_res["admin"], "game_id": redis_res["game_id"],
"players": redis_res["players"], "answers": []}))
await sio.emit("player_joined", {"username": data["username"], "sid": sid}, room=redis_res["admin"])
sio.enter_room(sid, data["game_pin"]) # TODO: make more secure
@sio.event
async def start_game(sid, data):
print(sid, data, "START_GAME")
session = await sio.get_session(sid)
print(session)
if session["admin"]:
await sio.emit("start_game", room=session["game_pin"])
@sio.event
async def register_as_admin(sid, data):
game_pin = data["game_pin"]
game_id = data["game_id"]
if (await redis.get(f"game_session:{game_pin}")) is None:
await redis.set(f"game_session:{game_pin}", json.dumps({"admin": sid, "game_id": game_id, "players": []}))
await sio.emit("registered_as_admin", {"game_id": game_id, "game": await redis.get(f"game:{data['game_pin']}")},
room=sid)
async with sio.session(sid) as session:
session['game_pin'] = data["game_pin"]
session["admin"] = True
sio.enter_room(sid, data["game_pin"])
else:
await sio.emit("already_registered_as_admin", room=sid)
@sio.event
async def get_question_results(sid, data):
session = await sio.get_session(sid)
if session["admin"]:
redis_res = await redis.get(f"game_session:{session['game_pin']}:{data['question_number']}")
game_pin = session['game_pin']
await sio.emit("question_results", redis_res, room=game_pin)
@sio.event
async def set_question_number(sid, data):
session = await sio.get_session(sid)
if session["admin"]:
game_pin = session['game_pin']
await sio.emit("set_question_number", data, room=game_pin)
@sio.event
async def submit_answer(sid, data):
session = await sio.get_session(sid)
redis_res = await redis.get(f"game_session:{session['game_pin']}")
game_session = GameSession(**json.loads(redis_res))
game_data = PlayGame(**json.loads(await redis.get(f"game:{session['game_pin']}")))
print(game_session)
answer_right = False
for answer in game_data.questions[int(data["question_index"])].answers:
if answer.answer == data["answer"] and answer.right:
answer_right = True
break
answers = await redis.get(f"game_session:{session['game_pin']}:{data['question_index']}")
if answers is None:
await redis.set(f"game_session:{session['game_pin']}:{data['question_index']}",
json.dumps(
[{"username": session["username"], "answer": data["answer"], "right": answer_right}]))
else:
answers = json.loads(answers)
answers.append({session["username"]: data["answer"]})
await redis.set(f"game_session:{session['game_pin']}:{data['question_index']}", json.dumps(answers))
# await redis.set(f"game_data:{session['game_pin']}", json.dumps(data))
# @sio.event
# async def admin_game(sid, data):
# redis_res = await redis.get(f"game:{data['game_pin']}")
# if redis_res is None:
# await sio.emit("game_not_found", room=sid)
# else:
#
# await sio.emit("joined_game", data, room=data['game_pin'])
# @sio.event
# async def start_game(sid, data):
# game_pin = (await sio.get_session(sid))['game_pin']
# session = await sio.get_session(sid)
# if session['admin']:
# await sio.emit("start_game", data, room=game_pin)
@sio.event
async def get_game_data(sid, data):
game_pin = (await sio.get_session(sid))['game_pin']
game_data = await redis.get(f"game:{game_pin}")
if game_data is not None:
await sio.emit("game_data", json.loads(game_data), room=game_pin)
print(sid, data, "GET_GAME_DATA")
#
# @sio.event
# async def message(sid, data):
# game_pin = (await sio.get_session(sid))['game_pin']
# await sio.emit("message", data, room=game_pin)
#
#
# @sio.on('*')
# async def catch_all(event, sid, data):
# print(event, sid, data)