Committing Everything
This commit is contained in:
BogPlaymate
Mawoka
@@ -0,0 +1,107 @@
|
||||
# SPDX-FileCopyrightText: 2023 Marlon W (Mawoka)
|
||||
#
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
import re
|
||||
from functools import lru_cache
|
||||
|
||||
from redis import asyncio as redis_lib
|
||||
import redis as redis_base_lib
|
||||
from pydantic import BaseSettings, RedisDsn, PostgresDsn, BaseModel
|
||||
import meilisearch as MeiliSearch
|
||||
from typing import Optional
|
||||
from arq import create_pool
|
||||
from arq.connections import RedisSettings, ArqRedis
|
||||
|
||||
from classquiz.storage import Storage
|
||||
|
||||
|
||||
class CustomOpenIDProvider(BaseModel):
|
||||
scopes: str = "openid email profile"
|
||||
server_metadata_url: str
|
||||
client_id: str
|
||||
client_secret: str
|
||||
|
||||
|
||||
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 | PostgresDsn = "postgresql://postgres:mysecretpassword@localhost:5432/classquiz"
|
||||
hcaptcha_key: str | None = None
|
||||
recaptcha_key: str | None = None
|
||||
mail_address: str
|
||||
mail_password: str
|
||||
mail_username: str
|
||||
mail_server: str
|
||||
mail_port: int
|
||||
secret_key: str
|
||||
access_token_expire_minutes: int = 30
|
||||
cache_expiry: int = 86400
|
||||
sentry_dsn: str | None
|
||||
meilisearch_url: str = "http://127.0.0.1:7700"
|
||||
meilisearch_index: str = "classquiz"
|
||||
google_client_id: Optional[str]
|
||||
google_client_secret: Optional[str]
|
||||
github_client_id: Optional[str]
|
||||
github_client_secret: Optional[str]
|
||||
custom_openid_provider: CustomOpenIDProvider | None = None
|
||||
telemetry_enabled: bool = True
|
||||
free_storage_limit: int = 1074000000
|
||||
pixabay_api_key: str | None = None
|
||||
mods: list[str] = []
|
||||
registration_disabled: bool = False
|
||||
|
||||
# storage_backend
|
||||
storage_backend: str | None = "local"
|
||||
|
||||
# if storage_backend == "local":
|
||||
storage_path: str | None
|
||||
|
||||
# if storage_backend == "s3":
|
||||
s3_access_key: str | None
|
||||
s3_secret_key: str | None
|
||||
s3_bucket_name: str = "classquiz"
|
||||
s3_base_url: str | None
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
env_file_encoding = "utf-8"
|
||||
env_nested_delimiter = "__"
|
||||
|
||||
|
||||
async def initialize_arq():
|
||||
# skipcq: PYL-W0603
|
||||
global arq
|
||||
arq = await create_pool(RedisSettings.from_dsn(settings.redis))
|
||||
|
||||
|
||||
@lru_cache()
|
||||
def settings() -> Settings:
|
||||
return Settings()
|
||||
|
||||
|
||||
pool = redis_lib.ConnectionPool().from_url(settings().redis)
|
||||
|
||||
redis: redis_base_lib.client.Redis = redis_lib.Redis(connection_pool=pool)
|
||||
arq: ArqRedis = ArqRedis(pool_or_conn=pool)
|
||||
storage: Storage = Storage(
|
||||
backend=settings().storage_backend,
|
||||
storage_path=settings().storage_path,
|
||||
access_key=settings().s3_access_key,
|
||||
secret_key=settings().s3_secret_key,
|
||||
bucket_name=settings().s3_bucket_name,
|
||||
base_url=settings().s3_base_url,
|
||||
)
|
||||
|
||||
meilisearch = MeiliSearch.Client(settings().meilisearch_url)
|
||||
|
||||
ALLOWED_TAGS_FOR_QUIZ = ["b", "strong", "i", "em", "small", "mark", "del", "sub", "sup"]
|
||||
|
||||
ALLOWED_MIME_TYPES = ["image/png", "video/mp4", "image/jpeg", "image/gif", "image/webp"]
|
||||
|
||||
server_regex = rf"^{re.escape(settings().root_address)}/api/v1/storage/download/.{{36}}--.{{36}}$"
|
||||
+22
-19
@@ -7,7 +7,8 @@ from functools import lru_cache
|
||||
|
||||
from redis import asyncio as redis_lib
|
||||
import redis as redis_base_lib
|
||||
from pydantic import BaseSettings, RedisDsn, PostgresDsn, BaseModel
|
||||
from pydantic_settings import BaseSettings
|
||||
from pydantic import RedisDsn, PostgresDsn, BaseModel
|
||||
import meilisearch as MeiliSearch
|
||||
from typing import Optional
|
||||
from arq import create_pool
|
||||
@@ -32,23 +33,25 @@ class Settings(BaseSettings):
|
||||
redis: RedisDsn = "redis://localhost:6379/0?decode_responses=True"
|
||||
skip_email_verification: bool = False
|
||||
db_url: str | PostgresDsn = "postgresql://postgres:mysecretpassword@localhost:5432/classquiz"
|
||||
hcaptcha_key: str | None = None
|
||||
hcaptcha_key: str = "3852831a-48b4-4236-b0de-9769a1998468"
|
||||
recaptcha_key: str | None = None
|
||||
mail_address: str
|
||||
mail_password: str
|
||||
mail_username: str
|
||||
mail_server: str
|
||||
mail_port: int
|
||||
secret_key: str
|
||||
# mail_address: str
|
||||
mail_address: str = "778@madeupaddress.com"
|
||||
# mail_password: str
|
||||
mail_password: str = "mail_passwordstring"
|
||||
mail_username: str = "mail_userstring"
|
||||
mail_server: str = "mail_serverstring"
|
||||
mail_port: int = 0
|
||||
secret_key: str = "secret_keystring"
|
||||
access_token_expire_minutes: int = 30
|
||||
cache_expiry: int = 86400
|
||||
sentry_dsn: str | None
|
||||
sentry_dsn: str = ""
|
||||
meilisearch_url: str = "http://127.0.0.1:7700"
|
||||
meilisearch_index: str = "classquiz"
|
||||
google_client_id: Optional[str]
|
||||
google_client_secret: Optional[str]
|
||||
github_client_id: Optional[str]
|
||||
github_client_secret: Optional[str]
|
||||
google_client_id: str = "google_client_idstring"
|
||||
google_client_secret: str = "google_client_secretstring"
|
||||
github_client_id: str = "github_client_idstring"
|
||||
github_client_secret: str = "github_client_secretstring"
|
||||
custom_openid_provider: CustomOpenIDProvider | None = None
|
||||
telemetry_enabled: bool = True
|
||||
free_storage_limit: int = 1074000000
|
||||
@@ -57,16 +60,16 @@ class Settings(BaseSettings):
|
||||
registration_disabled: bool = False
|
||||
|
||||
# storage_backend
|
||||
storage_backend: str | None = "local"
|
||||
storage_backend: str = "local"
|
||||
|
||||
# if storage_backend == "local":
|
||||
storage_path: str | None
|
||||
storage_path: str = "/tmp/storage"
|
||||
|
||||
# if storage_backend == "s3":
|
||||
s3_access_key: str | None
|
||||
s3_secret_key: str | None
|
||||
s3_access_key: str = "s3_access_keystring"
|
||||
s3_secret_key: str = "s3_secret_keystring"
|
||||
s3_bucket_name: str = "classquiz"
|
||||
s3_base_url: str | None
|
||||
s3_base_url: str = "s3_base_urlstring"
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
@@ -85,7 +88,7 @@ def settings() -> Settings:
|
||||
return Settings()
|
||||
|
||||
|
||||
pool = redis_lib.ConnectionPool().from_url(settings().redis)
|
||||
pool = redis_lib.ConnectionPool().from_url(str(settings().redis))
|
||||
|
||||
redis: redis_base_lib.client.Redis = redis_lib.Redis(connection_pool=pool)
|
||||
arq: ArqRedis = ArqRedis(pool_or_conn=pool)
|
||||
|
||||
@@ -0,0 +1,492 @@
|
||||
# SPDX-FileCopyrightText: 2023 Marlon W (Mawoka)
|
||||
#
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
import ormar
|
||||
from ormar import ReferentialAction
|
||||
from pydantic import BaseModel, Json, validator
|
||||
from enum import Enum
|
||||
from . import metadata, database
|
||||
from .quiztivity import QuizTivityPage
|
||||
from sqlalchemy import func
|
||||
|
||||
|
||||
class UserAuthTypes(Enum):
|
||||
LOCAL = "LOCAL"
|
||||
GOOGLE = "GOOGLE"
|
||||
GITHUB = "GITHUB"
|
||||
CUSTOM = "CUSTOM"
|
||||
|
||||
|
||||
class User(ormar.Model):
|
||||
"""
|
||||
The user model in the database
|
||||
"""
|
||||
|
||||
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: Optional[str] = ormar.String(max_length=100, nullable=True)
|
||||
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())
|
||||
auth_type: UserAuthTypes = ormar.Enum(enum_class=UserAuthTypes, default=UserAuthTypes.LOCAL)
|
||||
google_uid: Optional[str] = ormar.String(unique=True, max_length=255, nullable=True)
|
||||
avatar: bytes = ormar.LargeBinary(max_length=25000, represent_as_base64_str=True)
|
||||
github_user_id: int | None = ormar.Integer(nullable=True)
|
||||
require_password: bool = ormar.Boolean(default=True, nullable=False)
|
||||
backup_code: str = ormar.String(max_length=64, min_length=64, nullable=False, default=os.urandom(32).hex())
|
||||
totp_secret: str = ormar.String(max_length=32, min_length=32, nullable=True, default=None)
|
||||
storage_used: int = ormar.BigInteger(nullable=False, default=0, minimum=0)
|
||||
|
||||
class Meta:
|
||||
tablename = "users"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
class Config:
|
||||
use_enum_values = True
|
||||
|
||||
|
||||
class FidoCredentials(ormar.Model):
|
||||
pk: int = ormar.Integer(autoincrement=True, primary_key=True)
|
||||
id: bytes = ormar.LargeBinary(max_length=256)
|
||||
public_key: bytes = ormar.LargeBinary(max_length=256)
|
||||
sign_count: int = ormar.Integer()
|
||||
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
|
||||
class Meta:
|
||||
tablename = "fido_credentials"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class ApiKey(ormar.Model):
|
||||
key: str = ormar.String(max_length=48, min_length=48, primary_key=True)
|
||||
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
|
||||
class Meta:
|
||||
tablename = "api_keys"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class UserSession(ormar.Model):
|
||||
"""
|
||||
The user session model for user-sessions
|
||||
"""
|
||||
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4())
|
||||
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
session_key: str = ormar.String(unique=True, max_length=64)
|
||||
created_at: datetime = ormar.DateTime(default=datetime.now())
|
||||
ip_address: str = ormar.String(max_length=100, nullable=True)
|
||||
user_agent: str = ormar.String(max_length=255, nullable=True)
|
||||
last_seen: datetime = ormar.DateTime(default=datetime.now())
|
||||
|
||||
class Meta:
|
||||
tablename = "user_sessions"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class ABCDQuizAnswer(BaseModel):
|
||||
right: bool
|
||||
answer: str
|
||||
color: str | None
|
||||
|
||||
|
||||
class RangeQuizAnswer(BaseModel):
|
||||
min: int
|
||||
max: int
|
||||
min_correct: int
|
||||
max_correct: int
|
||||
|
||||
|
||||
class VotingQuizAnswer(BaseModel):
|
||||
answer: str
|
||||
image: str | None = None
|
||||
color: str | None
|
||||
|
||||
|
||||
class QuizQuestionType(str, Enum):
|
||||
ABCD = "ABCD"
|
||||
RANGE = "RANGE"
|
||||
VOTING = "VOTING"
|
||||
SLIDE = "SLIDE"
|
||||
TEXT = "TEXT"
|
||||
ORDER = "ORDER"
|
||||
CHECK = "CHECK"
|
||||
|
||||
|
||||
class TextQuizAnswer(BaseModel):
|
||||
answer: str
|
||||
case_sensitive: bool
|
||||
|
||||
|
||||
class QuizQuestion(BaseModel):
|
||||
question: str
|
||||
time: str # in Secs
|
||||
type: None | QuizQuestionType = QuizQuestionType.ABCD
|
||||
answers: list[ABCDQuizAnswer] | RangeQuizAnswer | list[TextQuizAnswer] | list[VotingQuizAnswer] | str
|
||||
image: str | None = None
|
||||
|
||||
@validator("answers")
|
||||
def answers_not_none_if_abcd_type(cls, v, values):
|
||||
if values["type"] == QuizQuestionType.ABCD and not isinstance(v[0], ABCDQuizAnswer):
|
||||
raise ValueError("Answers can't be none if type is ABCD")
|
||||
if values["type"] == QuizQuestionType.RANGE and not isinstance(v, RangeQuizAnswer):
|
||||
raise ValueError("Answer must be from type RangeQuizAnswer if type is RANGE")
|
||||
if values["type"] == QuizQuestionType.VOTING and not isinstance(v[0], VotingQuizAnswer):
|
||||
raise ValueError("Answer must be from type VotingQuizAnswer if type is VOTING")
|
||||
if values["type"] == QuizQuestionType.TEXT and not isinstance(v[0], TextQuizAnswer):
|
||||
raise ValueError("Answer must be from type TextQuizAnswer if type is TEXT")
|
||||
if values["type"] == QuizQuestionType.ORDER and not isinstance(v[0], VotingQuizAnswer):
|
||||
raise ValueError("Answer must be from type VotingQuizAnswer if type is ORDER")
|
||||
if values["type"] == QuizQuestionType.SLIDE and not isinstance(v, str):
|
||||
raise ValueError("Answer must be from type SlideElement if type is SLIDE")
|
||||
if values["type"] == QuizQuestionType.CHECK and not isinstance(v[0], ABCDQuizAnswer):
|
||||
raise ValueError("Answers can't be none if type is CHECK")
|
||||
return v
|
||||
|
||||
|
||||
class QuizInput(BaseModel):
|
||||
public: bool = False
|
||||
title: str
|
||||
description: str
|
||||
cover_image: str | None
|
||||
background_color: str | None
|
||||
questions: list[QuizQuestion]
|
||||
background_image: str | None
|
||||
|
||||
|
||||
class Quiz(ormar.Model):
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4(), nullable=False, unique=True)
|
||||
public: bool = ormar.Boolean(default=False)
|
||||
title: str = ormar.Text()
|
||||
description: str = ormar.Text(nullable=True)
|
||||
created_at: datetime = ormar.DateTime(default=datetime.now())
|
||||
updated_at: datetime = ormar.DateTime(default=datetime.now())
|
||||
user_id: uuid.UUID = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
questions: Json[list[QuizQuestion]] = ormar.JSON(nullable=False)
|
||||
imported_from_kahoot: Optional[bool] = ormar.Boolean(default=False, nullable=True)
|
||||
cover_image: Optional[str] = ormar.Text(nullable=True, unique=False)
|
||||
background_color: str | None = ormar.Text(nullable=True, unique=False)
|
||||
background_image: str | None = ormar.Text(nullable=True, unique=False)
|
||||
kahoot_id: uuid.UUID | None = ormar.UUID(nullable=True, default=None)
|
||||
likes: int = ormar.Integer(nullable=False, default=0, server_default="0")
|
||||
dislikes: int = ormar.Integer(nullable=False, default=0, server_default="0")
|
||||
plays: int = ormar.Integer(nullable=False, default=0, server_default="0")
|
||||
views: int = ormar.Integer(nullable=False, default=0, server_default="0")
|
||||
mod_rating: int | None = ormar.SmallInteger(nullable=True)
|
||||
|
||||
class Meta:
|
||||
tablename = "quiz"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class InstanceData(ormar.Model):
|
||||
instance_id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4(), nullable=False, unique=True)
|
||||
|
||||
class Meta:
|
||||
tablename = "instance_data"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class Token(BaseModel):
|
||||
"""
|
||||
For JWT
|
||||
"""
|
||||
|
||||
access_token: str
|
||||
token_type: str
|
||||
|
||||
|
||||
class TokenData(BaseModel):
|
||||
"""
|
||||
For JWT
|
||||
"""
|
||||
|
||||
email: str | None = None
|
||||
|
||||
|
||||
class PlayGame(BaseModel):
|
||||
quiz_id: uuid.UUID | str
|
||||
description: str
|
||||
user_id: uuid.UUID
|
||||
title: str
|
||||
questions: list[QuizQuestion]
|
||||
game_id: uuid.UUID
|
||||
game_pin: str
|
||||
started: bool = False
|
||||
captcha_enabled: bool = False
|
||||
cover_image: str | None
|
||||
game_mode: str | None
|
||||
current_question: int = -1
|
||||
background_color: str | None
|
||||
background_image: str | None
|
||||
custom_field: str | None
|
||||
question_show: bool = False
|
||||
|
||||
|
||||
class GamePlayer(BaseModel):
|
||||
username: str
|
||||
sid: str | None
|
||||
|
||||
|
||||
class GameAnswer2(BaseModel):
|
||||
username: str
|
||||
right: bool
|
||||
answer: str
|
||||
|
||||
|
||||
class GameAnswer1(BaseModel):
|
||||
id: int
|
||||
answers: list[GameAnswer2]
|
||||
|
||||
|
||||
class GameSession(BaseModel):
|
||||
admin: str
|
||||
game_id: str
|
||||
# players: list[GamePlayer | None]
|
||||
answers: list[GameAnswer1 | None]
|
||||
|
||||
|
||||
class UpdatePassword(BaseModel):
|
||||
old_password: str
|
||||
new_password: str
|
||||
|
||||
|
||||
class AnswerData(BaseModel):
|
||||
username: str
|
||||
answer: str
|
||||
right: bool
|
||||
time_taken: float # In milliseconds
|
||||
score: int
|
||||
|
||||
|
||||
class AnswerDataList(BaseModel):
|
||||
# Just a method to make a top-level list
|
||||
__root__: list[AnswerData]
|
||||
|
||||
|
||||
class GameInLobby(BaseModel):
|
||||
game_pin: str
|
||||
quiz_title: str
|
||||
game_id: uuid.UUID
|
||||
|
||||
|
||||
# skipcq: PY-W0069
|
||||
# class UserProfileLinks(ormar.Model):
|
||||
# id: int = ormar.Integer(primary_key=True, autoincrement=True)
|
||||
# user: Optional[User] = ormar.ForeignKey(User)
|
||||
# github_username: str | None = ormar.Text(nullable=True)
|
||||
# reddit_username: str | None = ormar.Text(nullable=True)
|
||||
# kahoot_user_id: str | None = ormar.Text(nullable=True)
|
||||
|
||||
|
||||
class GameResults(ormar.Model):
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True)
|
||||
quiz: uuid.UUID | Quiz = ormar.ForeignKey(Quiz, ondelete=ReferentialAction.CASCADE)
|
||||
user: uuid.UUID | User = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
timestamp: datetime = ormar.DateTime(default=datetime.now(), nullable=False)
|
||||
player_count: int = ormar.Integer(nullable=False, default=0)
|
||||
note: str | None = ormar.Text(nullable=True)
|
||||
answers: Json[list[AnswerData]] = ormar.JSON(True)
|
||||
player_scores: Json[dict[str, str]] = ormar.JSON(nullable=True)
|
||||
custom_field_data: Json[dict[str, str]] | None = ormar.JSON(nullable=True)
|
||||
title: str = ormar.Text(nullable=False)
|
||||
description: str = ormar.Text(nullable=False)
|
||||
questions: Json[list[QuizQuestion]] = ormar.JSON(nullable=False)
|
||||
|
||||
class Meta:
|
||||
tablename = "game_results"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class QuizTivityInput(BaseModel):
|
||||
title: str
|
||||
pages: list[QuizTivityPage]
|
||||
|
||||
|
||||
class QuizTivity(ormar.Model):
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True)
|
||||
title: str = ormar.Text(nullable=False)
|
||||
created_at: datetime = ormar.DateTime(nullable=False, server_default=func.now())
|
||||
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
pages: list[QuizTivityPage] = ormar.JSON(nullable=False)
|
||||
|
||||
class Meta:
|
||||
tablename = "quiztivitys"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class QuizTivityShare(ormar.Model):
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True)
|
||||
name: str | None = ormar.Text(nullable=True)
|
||||
expire_at: datetime | None = ormar.DateTime(nullable=True)
|
||||
quiztivity: QuizTivity | None = ormar.ForeignKey(QuizTivity, ondelete=ReferentialAction.CASCADE)
|
||||
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
|
||||
class Meta:
|
||||
tablename = "quiztivityshares"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class OnlyId(BaseModel):
|
||||
id: uuid.UUID
|
||||
|
||||
|
||||
class PublicQuizTivityShare(BaseModel):
|
||||
id: uuid.UUID
|
||||
name: str | None
|
||||
expire_in: int | None
|
||||
quiztivity: OnlyId
|
||||
user: OnlyId
|
||||
|
||||
@classmethod
|
||||
def from_db_model(cls, data: QuizTivityShare):
|
||||
expire_in = None
|
||||
if data.expire_at is not None:
|
||||
expire_in = int((data.expire_at - datetime.now()).seconds / 60)
|
||||
return cls(
|
||||
id=data.id,
|
||||
name=data.name,
|
||||
expire_in=expire_in,
|
||||
quiztivity=OnlyId(id=data.quiztivity.id),
|
||||
user=OnlyId(id=data.user.id),
|
||||
)
|
||||
|
||||
|
||||
class StorageItem(ormar.Model):
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True)
|
||||
uploaded_at: datetime = ormar.DateTime(nullable=False, default=datetime.now())
|
||||
mime_type: str = ormar.Text(nullable=False)
|
||||
hash: bytes | None = ormar.LargeBinary(nullable=True, min_length=16, max_length=16)
|
||||
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.SET_NULL)
|
||||
size: int = ormar.BigInteger(nullable=False)
|
||||
storage_path: str | None = ormar.Text(nullable=True)
|
||||
deleted_at: datetime | None = ormar.DateTime(nullable=True, default=None)
|
||||
quiztivities: list[QuizTivity] | None = ormar.ManyToMany(QuizTivity)
|
||||
quizzes: list[Quiz] | None = ormar.ManyToMany(Quiz)
|
||||
alt_text: str | None = ormar.Text(default=None, nullable=True)
|
||||
filename: str | None = ormar.Text(default=None, nullable=True)
|
||||
thumbhash: str | None = ormar.Text(default=None, nullable=True)
|
||||
server: str | None = ormar.Text(default=None, nullable=True)
|
||||
imported: bool = ormar.Boolean(default=False, nullable=True)
|
||||
|
||||
class Meta:
|
||||
tablename = "storage_items"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class PublicStorageItem(BaseModel):
|
||||
id: uuid.UUID
|
||||
uploaded_at: datetime
|
||||
mime_type: str
|
||||
hash: str | None
|
||||
size: int
|
||||
deleted_at: datetime | None
|
||||
alt_text: str | None
|
||||
filename: str | None
|
||||
thumbhash: str | None
|
||||
server: str | None
|
||||
imported: bool
|
||||
|
||||
@classmethod
|
||||
def from_db_model(cls, data: StorageItem):
|
||||
hash_data = None
|
||||
if data.hash is not None:
|
||||
hash_data = data.hash.hex()
|
||||
return cls(
|
||||
id=data.id,
|
||||
uploaded_at=data.uploaded_at,
|
||||
mime_type=data.mime_type,
|
||||
hash=hash_data,
|
||||
size=data.size,
|
||||
deleted_at=data.deleted_at,
|
||||
alt_text=data.alt_text,
|
||||
filename=data.filename,
|
||||
thumbhash=data.thumbhash,
|
||||
server=data.server,
|
||||
imported=data.imported,
|
||||
)
|
||||
|
||||
|
||||
class PrivateStorageItem(PublicStorageItem):
|
||||
quizzes: list[OnlyId]
|
||||
quiztivities: list[OnlyId]
|
||||
|
||||
@classmethod
|
||||
def from_db_model(cls, data: StorageItem):
|
||||
hash_data = None
|
||||
if data.hash is not None:
|
||||
hash_data = data.hash.hex()
|
||||
quiztivities = []
|
||||
quizzes = []
|
||||
for quiz in data.quizzes:
|
||||
quizzes.append(OnlyId(id=quiz.id))
|
||||
for quiztivity in data.quiztivities:
|
||||
quiztivities.append(OnlyId(id=quiztivity.id))
|
||||
return cls(
|
||||
id=data.id,
|
||||
uploaded_at=data.uploaded_at,
|
||||
mime_type=data.mime_type,
|
||||
hash=hash_data,
|
||||
size=data.size,
|
||||
deleted_at=data.deleted_at,
|
||||
alt_text=data.alt_text,
|
||||
filename=data.filename,
|
||||
quiztivities=quiztivities,
|
||||
quizzes=quizzes,
|
||||
thumbhash=data.thumbhash,
|
||||
server=data.server,
|
||||
imported=data.imported,
|
||||
)
|
||||
|
||||
|
||||
class UpdateStorageItem(BaseModel):
|
||||
filename: str | None
|
||||
alt_text: str | None
|
||||
|
||||
|
||||
class Controller(ormar.Model):
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True)
|
||||
user: uuid.UUID | User = ormar.ForeignKey(User)
|
||||
secret_key: str = ormar.String(nullable=False, max_length=24, min_length=24)
|
||||
player_name: str = ormar.Text(nullable=False)
|
||||
last_seen: datetime | None = ormar.DateTime(nullable=True)
|
||||
first_seen: datetime | None = ormar.DateTime(nullable=True)
|
||||
name: str = ormar.Text(nullable=False)
|
||||
os_version: str | None = ormar.Text(nullable=True)
|
||||
wanted_os_version: str = ormar.Text(nullable=True, default=None)
|
||||
|
||||
class Meta:
|
||||
tablename = "controller"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class Rating(ormar.Model):
|
||||
id: uuid.UUID = ormar.UUID(primary_key=True)
|
||||
user: uuid.UUID | User = ormar.ForeignKey(User)
|
||||
positive: bool = ormar.Boolean(nullable=False)
|
||||
created_at: datetime = ormar.DateTime(nullable=False, server_default=func.now())
|
||||
quiz: uuid.UUID | Quiz = ormar.ForeignKey(Quiz)
|
||||
|
||||
class Meta:
|
||||
tablename = "rating"
|
||||
metadata = metadata
|
||||
database = database
|
||||
+61
-48
@@ -45,13 +45,11 @@ class User(ormar.Model):
|
||||
totp_secret: str = ormar.String(max_length=32, min_length=32, nullable=True, default=None)
|
||||
storage_used: int = ormar.BigInteger(nullable=False, default=0, minimum=0)
|
||||
|
||||
class Meta:
|
||||
tablename = "users"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
class Config:
|
||||
use_enum_values = True
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="users"
|
||||
)
|
||||
|
||||
|
||||
class FidoCredentials(ormar.Model):
|
||||
@@ -61,20 +59,22 @@ class FidoCredentials(ormar.Model):
|
||||
sign_count: int = ormar.Integer()
|
||||
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
|
||||
class Meta:
|
||||
tablename = "fido_credentials"
|
||||
metadata = metadata
|
||||
database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="fido_credentials"
|
||||
)
|
||||
|
||||
|
||||
class ApiKey(ormar.Model):
|
||||
key: str = ormar.String(max_length=48, min_length=48, primary_key=True)
|
||||
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
|
||||
class Meta:
|
||||
tablename = "api_keys"
|
||||
metadata = metadata
|
||||
database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="api_keys"
|
||||
)
|
||||
|
||||
|
||||
class UserSession(ormar.Model):
|
||||
@@ -90,10 +90,11 @@ class UserSession(ormar.Model):
|
||||
user_agent: str = ormar.String(max_length=255, nullable=True)
|
||||
last_seen: datetime = ormar.DateTime(default=datetime.now())
|
||||
|
||||
class Meta:
|
||||
tablename = "user_sessions"
|
||||
metadata = metadata
|
||||
database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="user_sessions"
|
||||
)
|
||||
|
||||
|
||||
class ABCDQuizAnswer(BaseModel):
|
||||
@@ -186,19 +187,25 @@ class Quiz(ormar.Model):
|
||||
views: int = ormar.Integer(nullable=False, default=0, server_default="0")
|
||||
mod_rating: int | None = ormar.SmallInteger(nullable=True)
|
||||
|
||||
class Meta:
|
||||
tablename = "quiz"
|
||||
metadata = metadata
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
tablename = "quiz",
|
||||
metadata = metadata,
|
||||
database = database
|
||||
)
|
||||
|
||||
|
||||
class InstanceData(ormar.Model):
|
||||
instance_id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4(), nullable=False, unique=True)
|
||||
|
||||
class Meta:
|
||||
tablename = "instance_data"
|
||||
metadata = metadata
|
||||
# class Meta:
|
||||
# tablename = "instance_data"
|
||||
# metadata = metadata
|
||||
# database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
tablename = "instance_data",
|
||||
metadata = metadata,
|
||||
database = database
|
||||
)
|
||||
|
||||
|
||||
class Token(BaseModel):
|
||||
@@ -275,7 +282,7 @@ class AnswerData(BaseModel):
|
||||
|
||||
class AnswerDataList(BaseModel):
|
||||
# Just a method to make a top-level list
|
||||
__root__: list[AnswerData]
|
||||
root: list[AnswerData]
|
||||
|
||||
|
||||
class GameInLobby(BaseModel):
|
||||
@@ -307,10 +314,11 @@ class GameResults(ormar.Model):
|
||||
description: str = ormar.Text(nullable=False)
|
||||
questions: Json[list[QuizQuestion]] = ormar.JSON(nullable=False)
|
||||
|
||||
class Meta:
|
||||
tablename = "game_results"
|
||||
metadata = metadata
|
||||
database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="game_results"
|
||||
)
|
||||
|
||||
|
||||
class QuizTivityInput(BaseModel):
|
||||
@@ -325,10 +333,11 @@ class QuizTivity(ormar.Model):
|
||||
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
pages: list[QuizTivityPage] = ormar.JSON(nullable=False)
|
||||
|
||||
class Meta:
|
||||
tablename = "quiztivitys"
|
||||
metadata = metadata
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
tablename = "quiztivitys",
|
||||
metadata = metadata,
|
||||
database = database
|
||||
)
|
||||
|
||||
|
||||
class QuizTivityShare(ormar.Model):
|
||||
@@ -338,10 +347,11 @@ class QuizTivityShare(ormar.Model):
|
||||
quiztivity: QuizTivity | None = ormar.ForeignKey(QuizTivity, ondelete=ReferentialAction.CASCADE)
|
||||
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
|
||||
|
||||
class Meta:
|
||||
tablename = "quiztivityshares"
|
||||
metadata = metadata
|
||||
database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="quiztivityshares"
|
||||
)
|
||||
|
||||
|
||||
class OnlyId(BaseModel):
|
||||
@@ -386,10 +396,11 @@ class StorageItem(ormar.Model):
|
||||
server: str | None = ormar.Text(default=None, nullable=True)
|
||||
imported: bool = ormar.Boolean(default=False, nullable=True)
|
||||
|
||||
class Meta:
|
||||
tablename = "storage_items"
|
||||
metadata = metadata
|
||||
database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="storage_items"
|
||||
)
|
||||
|
||||
|
||||
class PublicStorageItem(BaseModel):
|
||||
@@ -473,10 +484,11 @@ class Controller(ormar.Model):
|
||||
os_version: str | None = ormar.Text(nullable=True)
|
||||
wanted_os_version: str = ormar.Text(nullable=True, default=None)
|
||||
|
||||
class Meta:
|
||||
tablename = "controller"
|
||||
metadata = metadata
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
tablename = "controller",
|
||||
metadata = metadata,
|
||||
database = database
|
||||
)
|
||||
|
||||
|
||||
class Rating(ormar.Model):
|
||||
@@ -486,7 +498,8 @@ class Rating(ormar.Model):
|
||||
created_at: datetime = ormar.DateTime(nullable=False, server_default=func.now())
|
||||
quiz: uuid.UUID | Quiz = ormar.ForeignKey(Quiz)
|
||||
|
||||
class Meta:
|
||||
tablename = "rating"
|
||||
metadata = metadata
|
||||
database = database
|
||||
ormar_config = ormar.OrmarConfig(
|
||||
database=database,
|
||||
metadata=metadata,
|
||||
tablename="rating"
|
||||
)
|
||||
|
||||
@@ -102,19 +102,27 @@ def verify_webauthn(data, fidocredentialss: list[FidoCredentials], login_session
|
||||
@router.post("/start")
|
||||
async def start_login(data: StartLoginInput):
|
||||
user = (
|
||||
#TODO Find out why the fidocredentialls call fails
|
||||
await User.objects.select_related("fidocredentialss")
|
||||
.filter((User.email == data.email) | (User.username == data.email))
|
||||
.get_or_none()
|
||||
# await User.objects.filter((User.email == data.email) | (User.username == data.email))
|
||||
# .get_or_none()
|
||||
)
|
||||
print("User.objects.filter call completed...")
|
||||
step_1: set[StartLoginResponseTypes] = set()
|
||||
step_2: set[StartLoginResponseTypes] = set()
|
||||
webauthn_data = None
|
||||
webauthn_challenge = None
|
||||
print("\"webauthn_challenge = None\" call completed...")
|
||||
print("user.verified: " + str(user.verified))
|
||||
if user is None or not user.verified:
|
||||
step_1.add(StartLoginResponseTypes.PASSWORD)
|
||||
return StartLoginResponse(step_1=step_1, step_2=step_2, session_id=os.urandom(16).hex(), webauthn_data=None)
|
||||
print("Okay. User is not None or unverified...")
|
||||
if user.password is not None:
|
||||
step_1.add(StartLoginResponseTypes.PASSWORD)
|
||||
''' TODO Find out why fidocredentialss is causing problems
|
||||
if len(user.fidocredentialss) > 0:
|
||||
if user.require_password is True:
|
||||
step_2.add(StartLoginResponseTypes.PASSKEY)
|
||||
@@ -129,6 +137,8 @@ async def start_login(data: StartLoginInput):
|
||||
)
|
||||
webauthn_challenge = base64.b64encode(webauthn_data.challenge).decode("utf-8")
|
||||
webauthn_data = options_to_json(webauthn_data)
|
||||
'''
|
||||
print("Moving past commented code, the long string one...")
|
||||
if user.totp_secret is not None:
|
||||
if user.require_password:
|
||||
step_2.add(StartLoginResponseTypes.TOTP)
|
||||
@@ -170,7 +180,8 @@ async def step_1_endpoint(session_id: str, data: StepInput, request: Request, re
|
||||
else:
|
||||
print("unknown step")
|
||||
raise HTTPException(401)
|
||||
user = await User.objects.select_related("fidocredentialss").get_or_none(id=uuid.UUID(login_session.user_id))
|
||||
#TODO fidocredentials bug fix user = await User.objects.select_related("fidocredentialss").get_or_none(id=uuid.UUID(login_session.user_id))
|
||||
user = await User.objects.get_or_none(id=uuid.UUID(login_session.user_id))
|
||||
if data.auth_type == StartLoginResponseTypes.PASSWORD:
|
||||
if verify_password(data.data, user.password):
|
||||
if len(login_session.step_2) == 0 or (step_id == 2 and login_session.step1_success is True):
|
||||
|
||||
@@ -15,7 +15,7 @@ import ormar.exceptions
|
||||
from classquiz.helpers import generate_spreadsheet, handle_import_from_excel
|
||||
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
|
||||
from fastapi.responses import JSONResponse, StreamingResponse
|
||||
from pydantic import ValidationError, BaseModel
|
||||
from pydantic import ValidationError, BaseModel, Field
|
||||
|
||||
from classquiz.auth import get_current_user
|
||||
from classquiz.config import redis, settings, storage, meilisearch
|
||||
@@ -56,7 +56,8 @@ class PublicQuizResponseUser(BaseModel):
|
||||
|
||||
class PublicQuizResponse(Quiz.get_pydantic()):
|
||||
user_id: PublicQuizResponseUser
|
||||
questions: list[QuizQuestion]
|
||||
# questions: list[QuizQuestion]
|
||||
var_questions: list[QuizQuestion] = Field(..., alias='questions')
|
||||
likes: int
|
||||
dislikes: int
|
||||
views: int
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
# SPDX-FileCopyrightText: 2023 Marlon W (Mawoka)
|
||||
#
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
|
||||
import datetime
|
||||
import uuid
|
||||
|
||||
from fastapi import APIRouter, Response
|
||||
from jinja2 import Template
|
||||
from pydantic import BaseModel
|
||||
from classquiz.config import redis, settings
|
||||
from classquiz.db import database
|
||||
|
||||
settings = settings()
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
sitemap_template = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
|
||||
{% for entry in pages -%}
|
||||
<url>
|
||||
<loc>{{ root_address}}/view/{{ entry.id.hex }}</loc>
|
||||
{%- if entry.updated_at != None -%}
|
||||
<lastmod>{{ entry.updated_at.strftime("%Y-%m-%d") }}</lastmod>
|
||||
{% endif %}
|
||||
</url>
|
||||
{%- endfor %}
|
||||
</urlset>"""
|
||||
|
||||
|
||||
sql_statement_metadata = """
|
||||
SELECT id, title, description, updated_at from quiz where public ='t'
|
||||
"""
|
||||
|
||||
sql_statement_id_and_modified_only = """
|
||||
SELECT id, updated_at from quiz where public ='t'
|
||||
"""
|
||||
|
||||
template = Template(sitemap_template, enable_async=True)
|
||||
|
||||
|
||||
class SitemapQuiz(BaseModel):
|
||||
id: uuid.UUID
|
||||
updated_at: datetime.datetime
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
|
||||
@router.get("/get")
|
||||
async def get_sitemap():
|
||||
redis_cache_resp = await redis.get("sitemap")
|
||||
if redis_cache_resp is None:
|
||||
res = await database.fetch_all(sql_statement_id_and_modified_only)
|
||||
entries = []
|
||||
for i in res:
|
||||
entries.append(SitemapQuiz.from_orm(i))
|
||||
rendered_sitemap = await template.render_async({"pages": entries, "root_address": settings.root_address})
|
||||
await redis.set("sitemap", rendered_sitemap, ex=86400)
|
||||
return Response(content=rendered_sitemap, media_type="application/xml")
|
||||
else:
|
||||
return Response(content=redis_cache_resp, media_type="application/xml")
|
||||
@@ -45,7 +45,7 @@ class SitemapQuiz(BaseModel):
|
||||
updated_at: datetime.datetime
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
from_attributes = True
|
||||
|
||||
|
||||
@router.get("/get")
|
||||
|
||||
@@ -27,4 +27,4 @@ class WorkerSettings:
|
||||
cron_jobs = [cron(clean_editor_images_up, hour={0, 6, 12, 18}, minute=0)]
|
||||
on_startup = startup
|
||||
on_shutdown = shutdown
|
||||
redis_settings = RedisSettings.from_dsn(settings.redis)
|
||||
redis_settings = RedisSettings.from_dsn(str(settings.redis))
|
||||
|
||||
Reference in New Issue
Block a user