Add pre-commit

This commit is contained in:
Matthew Welch 2023-11-27 19:25:33 -08:00
parent ac9b165ca0
commit d131482dd8
19 changed files with 311 additions and 169 deletions

2
.flake8 Normal file
View File

@ -0,0 +1,2 @@
[flake8]
max_line_length = 150

View File

@ -4,4 +4,4 @@ venv/
.gcloudignore .gcloudignore
.git .git
.gitignore .gitignore
*.tsv *.tsv

43
.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,43 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: name-tests-test
- id: requirements-txt-fixer
- repo: https://github.com/asottile/setup-cfg-fmt
rev: v2.5.0
hooks:
- id: setup-cfg-fmt
- repo: https://github.com/asottile/reorder_python_imports
rev: v3.12.0
hooks:
- id: reorder-python-imports
args: [--py37-plus, --add-import, 'from __future__ import annotations']
- repo: https://github.com/asottile/add-trailing-comma
rev: v3.1.0
hooks:
- id: add-trailing-comma
- repo: https://github.com/asottile/pyupgrade
rev: v3.15.0
hooks:
- id: pyupgrade
- repo: https://github.com/pre-commit/mirrors-autopep8
rev: v2.0.4
hooks:
- id: autopep8
- repo: https://github.com/PyCQA/flake8
rev: 6.1.0
hooks:
- id: flake8
additional_dependencies:
- flake8-use-pathlib
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.7.0
hooks:
- id: mypy
additional_dependencies:
- types-toml
- types-requests

View File

@ -1,6 +1,15 @@
from __future__ import annotations
import json import json
from flask import render_template, Blueprint, request, jsonify, redirect, url_for
from flask_security import roles_required, permissions_accepted from flask import Blueprint
from flask import jsonify
from flask import redirect
from flask import render_template
from flask import request
from flask import url_for
from flask_security import permissions_accepted
from QuizTheWord import database from QuizTheWord import database
@ -116,12 +125,12 @@ def query_users():
def parse_question_query(query): def parse_question_query(query):
if query: if query:
query: dict = json.loads(query) parsed_query: dict = json.loads(query)
if "multiple_choice" in query.keys(): if "multiple_choice" in parsed_query.keys():
query["multiple_choice"] = True if query["multiple_choice"] == "true" else False parsed_query["multiple_choice"] = True if parsed_query["multiple_choice"] == "true" else False
if "hidden_answer" in query.keys(): if "hidden_answer" in parsed_query.keys():
query["hidden_answer"] = True if query["hidden_answer"] == "true" else False parsed_query["hidden_answer"] = True if parsed_query["hidden_answer"] == "true" else False
return query return parsed_query
return None return None
@ -149,15 +158,15 @@ def parse_user_form_data(form):
data = { data = {
"username": form.get("username"), "username": form.get("username"),
"email": form.get("email"), "email": form.get("email"),
"roles": form.getlist("roles") "roles": form.getlist("roles"),
} }
return data return data
def parse_user_query(query): def parse_user_query(query):
if query: if query:
query: dict = json.loads(query) parsed_query: dict = json.loads(query)
# if "admin" in query.keys(): # if "admin" in query.keys():
# query["admin"] = True if query["admin"] == "true" else False # query["admin"] = True if query["admin"] == "true" else False
return query return parsed_query
return None return {}

View File

@ -7,4 +7,4 @@
* @license MIT * @license MIT
*/ */
@charset "UTF-8";.no-filter-control{height:34px}.filter-control{margin:0 2px 2px 2px}.ms-choice{border:0}.ms-parent>button:focus{outline:0} @charset "UTF-8";.no-filter-control{height:34px}.filter-control{margin:0 2px 2px 2px}.ms-choice{border:0}.ms-parent>button:focus{outline:0}

File diff suppressed because one or more lines are too long

View File

@ -45,4 +45,4 @@
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.0-beta3/dist/js/bootstrap.bundle.min.js" integrity="sha384-JEW9xMcG8R+pH31jmWH6WWP0WintQrMb4s7ZOdauHnUtxwoG2vI5DkLtS3qm9Ekf" crossorigin="anonymous"></script> <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.0-beta3/dist/js/bootstrap.bundle.min.js" integrity="sha384-JEW9xMcG8R+pH31jmWH6WWP0WintQrMb4s7ZOdauHnUtxwoG2vI5DkLtS3qm9Ekf" crossorigin="anonymous"></script>
{% block scripts %}{% endblock %} {% block scripts %}{% endblock %}
</body> </body>
</html> </html>

View File

@ -1,11 +1,25 @@
from __future__ import annotations
import os import os
import flask from authlib.integrations.flask_client import OAuth
from flask import request, render_template, jsonify, Flask, url_for, redirect, flash, session from flask import flash
from flask_security import Security, SQLAlchemySessionUserDatastore, login_user, logout_user, current_user from flask import Flask
from flask import jsonify
from flask import redirect
from flask import render_template
from flask import request
from flask import session
from flask import url_for
from flask_security import current_user
from flask_security import login_user
from flask_security import logout_user
from flask_security import Security
from flask_security import SQLAlchemySessionUserDatastore
from QuizTheWord import database from QuizTheWord import database
from QuizTheWord.admin import admin from QuizTheWord.admin import admin
from authlib.integrations.flask_client import OAuth
app = Flask(__name__) app = Flask(__name__)
environment_configuration = os.environ.get('CONFIGURATION_SETUP', "QuizTheWord.config.Development") environment_configuration = os.environ.get('CONFIGURATION_SETUP', "QuizTheWord.config.Development")
@ -19,8 +33,8 @@ with app.app_context():
name="google", name="google",
server_metadata_url=CONF_URL, server_metadata_url=CONF_URL,
client_kwargs={ client_kwargs={
"scope": "openid email profile" "scope": "openid email profile",
} },
) )
app.register_blueprint(admin.Admin) app.register_blueprint(admin.Admin)
@ -87,7 +101,7 @@ def check_answer():
def next_hidden_answer(): def next_hidden_answer():
if "hidden_answer_ids" not in session: if "hidden_answer_ids" not in session:
session["hidden_answer_ids"] = [] session["hidden_answer_ids"] = []
difficulty = request.args.get("difficulty", 1) difficulty = request.args.get("difficulty", 1, int)
next_question = database.next_hidden_answer(session["hidden_answer_ids"], difficulty) next_question = database.next_hidden_answer(session["hidden_answer_ids"], difficulty)
if next_question: if next_question:
session["hidden_answer_ids"].append(next_question.question_id) session["hidden_answer_ids"].append(next_question.question_id)
@ -102,7 +116,7 @@ def next_hidden_answer():
def next_multiple_choice(): def next_multiple_choice():
if "multiple_choice_ids" not in session: if "multiple_choice_ids" not in session:
session["multiple_choice_ids"] = [] session["multiple_choice_ids"] = []
difficulty = request.args.get("difficulty", 1) difficulty = request.args.get("difficulty", 1, int)
next_question = database.next_multiple_choice(session["multiple_choice_ids"], difficulty) next_question = database.next_multiple_choice(session["multiple_choice_ids"], difficulty)
if next_question: if next_question:
session["multiple_choice_ids"].append(next_question.question_id) session["multiple_choice_ids"].append(next_question.question_id)
@ -190,7 +204,7 @@ def error_404(e):
@app.errorhandler(500) @app.errorhandler(500)
def error_404(e): def error_500(e):
print(e) print(e)
msg = "There was an error with the server." msg = "There was an error with the server."
if app.config["DEBUG"]: if app.config["DEBUG"]:

View File

@ -1,4 +1,7 @@
from __future__ import annotations
from os import environ from os import environ
from dotenv import load_dotenv from dotenv import load_dotenv
@ -23,7 +26,7 @@ class Production(Config):
DEBUG = False DEBUG = False
DB_SOCKET_DIR = environ.get("DB_SOCKET_DIR", "/cloudsql") DB_SOCKET_DIR = environ.get("DB_SOCKET_DIR", "/cloudsql")
CLOUD_SQL_CONNECTION_NAME = environ.get("CLOUD_SQL_CONNECTION_NAME") CLOUD_SQL_CONNECTION_NAME = environ.get("CLOUD_SQL_CONNECTION_NAME")
DB_URL = f"postgresql+psycopg2://{Config.DB_USER}:{Config.DB_PASSWORD}@{Config.DB_HOST}:{Config.DB_PORT}/{Config.DB_NAME}?host={DB_SOCKET_DIR}/{CLOUD_SQL_CONNECTION_NAME}" DB_URL = f"postgresql+psycopg://{Config.DB_USER}:{Config.DB_PASSWORD}@{Config.DB_HOST}:{Config.DB_PORT}/{Config.DB_NAME}?host={DB_SOCKET_DIR}/{CLOUD_SQL_CONNECTION_NAME}" # noqa: E501
class Development(Config): class Development(Config):
@ -34,8 +37,8 @@ class Development(Config):
DB_PASSWORD = "hello" DB_PASSWORD = "hello"
DB_HOST = "localhost" DB_HOST = "localhost"
DB_NAME = "postgres" DB_NAME = "postgres"
DB_PORT = 5432 DB_PORT = "5432"
DB_URL = f"postgresql+psycopg2://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}" DB_URL = f"postgresql+psycopg://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
class Testing(Config): class Testing(Config):

View File

@ -1,34 +1,57 @@
from __future__ import annotations
import os import os
from flask import current_app, g
from flask_security import UserMixin, RoleMixin, SQLAlchemySessionUserDatastore, verify_password, hash_password
import sqlalchemy
from typing import Union, Optional, Literal, Type, List, Tuple
import random import random
from sqlalchemy import Column, JSON, String, Integer, create_engine, ForeignKey, func, Boolean, UnicodeText, DateTime, Numeric, CheckConstraint from datetime import datetime
from sqlalchemy.sql.expression import and_ from decimal import Decimal
from sqlalchemy.ext.declarative import declarative_base from typing import Any
from typing import Literal
from typing import Sequence
from flask import current_app
from flask import g
from flask_security import hash_password
from flask_security import RoleMixin
from flask_security import SQLAlchemySessionUserDatastore
from flask_security import UserMixin
from flask_security import verify_password
from sqlalchemy import and_
from sqlalchemy import CheckConstraint
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import JSON
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy.ext.associationproxy import association_proxy from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import sessionmaker, relationship, scoped_session from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
from werkzeug.utils import import_string from werkzeug.utils import import_string
config = import_string(os.environ.get("CONFIGURATION_SETUP")) config = import_string(os.environ.get("CONFIGURATION_SETUP"))
engine = create_engine(config.DB_URL, pool_size=20) engine = create_engine(config.DB_URL, pool_size=20)
Base = declarative_base()
def get_scoped_session(): class Base(DeclarativeBase):
pass
def get_scoped_session() -> scoped_session:
session_factory = sessionmaker(bind=engine) session_factory = sessionmaker(bind=engine)
return scoped_session(session_factory) return scoped_session(session_factory)
def get_session() -> sqlalchemy.orm.session.Session: def get_session() -> scoped_session:
if not hasattr(g, "session"): if not hasattr(g, "session"):
Session = get_scoped_session() session = get_scoped_session()
Base.query = Session.query_property() # This is for compatibility with Flask-Security-Too which assumes usage of Flask-Sqlalchemy # This is for compatibility with Flask-Security-Too which assumes usage of Flask-Sqlalchemy
# Session.query_property() Base.query = session.query_property()
g.session = Session() g.session = session()
return g.session return g.session
@ -48,18 +71,18 @@ def get_user_datastore() -> SQLAlchemySessionUserDatastore:
class User(Base, UserMixin): class User(Base, UserMixin):
__tablename__ = "users" __tablename__ = "users"
user_id = Column(Integer, primary_key=True) user_id: Mapped[int] = mapped_column(primary_key=True)
email = Column(String, unique=True, nullable=False) email: Mapped[str] = mapped_column(unique=True)
username = Column(String, unique=True) username: Mapped[str] = mapped_column(unique=True)
password = Column(String, nullable=True) password: Mapped[str | None]
active = Column(Boolean, nullable=False) active: Mapped[bool]
last_login_at = Column(DateTime()) last_login_at: Mapped[datetime]
current_login_at = Column(DateTime()) current_login_at: Mapped[datetime]
last_login_ip = Column(String(100)) last_login_ip: Mapped[str] = mapped_column(String(100))
current_login_ip = Column(String(100)) current_login_ip: Mapped[str] = mapped_column(String(100))
login_count = Column(Integer) login_count: Mapped[int]
fs_uniquifier = Column(String, unique=True, nullable=False) fs_uniquifier: Mapped[str] = mapped_column(unique=True)
google_id = Column(Numeric, nullable=True) google_id: Mapped[Decimal | None]
__table_args__ = ( __table_args__ = (
CheckConstraint("(password IS NOT NULL) OR (google_id IS NOT NULL)", "password_google_id_null_check"), CheckConstraint("(password IS NOT NULL) OR (google_id IS NOT NULL)", "password_google_id_null_check"),
@ -87,26 +110,26 @@ class User(Base, UserMixin):
class Role(Base, RoleMixin): class Role(Base, RoleMixin):
__tablename__ = "roles" __tablename__ = "roles"
role_id = Column(Integer, primary_key=True) role_id: Mapped[int] = mapped_column(primary_key=True)
name = Column(String, unique=True) name: Mapped[str] = mapped_column(unique=True)
description = Column(String) description: Mapped[str]
permissions = Column(UnicodeText) permissions: Mapped[str]
class UsersRoles(Base): class UsersRoles(Base):
__tablename__ = "users_roles" __tablename__ = "users_roles"
user_role_id = Column(Integer, primary_key=True) user_role_id: Mapped[int] = mapped_column(primary_key=True)
user_id = Column(Integer, ForeignKey("users.user_id")) user_id: Mapped[int] = mapped_column(ForeignKey("users.user_id"))
role_id = Column(Integer, ForeignKey("roles.role_id")) role_id: Mapped[int] = mapped_column(ForeignKey("roles.role_id"))
class AllQuestions(Base): class AllQuestions(Base):
__tablename__ = "all_questions" __tablename__ = "all_questions"
question_id = Column(Integer, primary_key=True, nullable=False) question_id: Mapped[int] = mapped_column(primary_key=True)
question_text = Column(String) question_text: Mapped[str]
answer = Column(String) answer: Mapped[str]
addresses = Column(String) addresses: Mapped[str]
multiple_choice = relationship("MultipleChoice", uselist=False, back_populates="all_question_relationship") multiple_choice = relationship("MultipleChoice", uselist=False, back_populates="all_question_relationship")
hidden_answer = relationship("HiddenAnswer", uselist=False, back_populates="all_question_relationship") hidden_answer = relationship("HiddenAnswer", uselist=False, back_populates="all_question_relationship")
@ -137,9 +160,9 @@ class AllQuestions(Base):
class HiddenAnswer(Base): class HiddenAnswer(Base):
__tablename__ = "category_hidden_answer" __tablename__ = "category_hidden_answer"
question_id = Column(Integer, ForeignKey("all_questions.question_id"), primary_key=True) question_id: Mapped[int] = mapped_column(ForeignKey("all_questions.question_id"), primary_key=True)
hidden_answer_difficulty = Column(Integer) hidden_answer_difficulty: Mapped[int]
hidden_answer_hint = Column(JSON) hidden_answer_hint: Mapped[dict | list | None] = mapped_column(JSON)
all_question_relationship = relationship("AllQuestions", lazy="joined", back_populates="hidden_answer") all_question_relationship = relationship("AllQuestions", lazy="joined", back_populates="hidden_answer")
question_text = association_proxy("all_question_relationship", "question_text") question_text = association_proxy("all_question_relationship", "question_text")
@ -169,10 +192,10 @@ class HiddenAnswer(Base):
class MultipleChoice(Base): class MultipleChoice(Base):
__tablename__ = "category_multiple_choice" __tablename__ = "category_multiple_choice"
question_id = Column(Integer, ForeignKey("all_questions.question_id"), primary_key=True) question_id: Mapped[int] = mapped_column(ForeignKey("all_questions.question_id"), primary_key=True)
multiple_choice_difficulty = Column(Integer) multiple_choice_difficulty: Mapped[int]
multiple_choice_hint = Column(JSON) multiple_choice_hint: Mapped[dict | list | None] = mapped_column(JSON)
wrong_answers = Column(JSON) wrong_answers: Mapped[list] = mapped_column(JSON)
all_question_relationship = relationship("AllQuestions", lazy="joined", back_populates="multiple_choice") all_question_relationship = relationship("AllQuestions", lazy="joined", back_populates="multiple_choice")
question_text = association_proxy("all_question_relationship", "question_text") question_text = association_proxy("all_question_relationship", "question_text")
@ -188,7 +211,7 @@ class MultipleChoice(Base):
self.answer_list = None self.answer_list = None
def randomize_answer_list(self): def randomize_answer_list(self):
answer_list: List[str] = [*self.wrong_answers, self.answer] answer_list: list[str] = [*self.wrong_answers, self.answer]
random.shuffle(answer_list) random.shuffle(answer_list)
self.answer_list = answer_list self.answer_list = answer_list
@ -220,7 +243,8 @@ class MultipleChoice(Base):
def add_multiple_choice_question(question, answer, addresses, difficulty, hint, wrong_answers): def add_multiple_choice_question(question, answer, addresses, difficulty, hint, wrong_answers):
session = get_session() session = get_session()
question_id = session.query(AllQuestions).order_by(AllQuestions.question_id.desc()).first().question_id + 1 select_stmt = select(AllQuestions).order_by(AllQuestions.question_id.desc())
question_id = session.scalars(select_stmt).first().question_id + 1
print(question_id) print(question_id)
base_question = AllQuestions(question_id, question, answer, addresses) base_question = AllQuestions(question_id, question, answer, addresses)
multiple_choice_question = MultipleChoice(question_id, difficulty, hint, wrong_answers, base_question) multiple_choice_question = MultipleChoice(question_id, difficulty, hint, wrong_answers, base_question)
@ -231,7 +255,8 @@ def add_multiple_choice_question(question, answer, addresses, difficulty, hint,
def add_question(data) -> AllQuestions: def add_question(data) -> AllQuestions:
session = get_session() session = get_session()
question_id = session.query(AllQuestions).order_by(AllQuestions.question_id.desc()).first().question_id + 1 select_stmt = select(AllQuestions).order_by(AllQuestions.question_id.desc())
question_id = session.scalars(select_stmt).first().question_id + 1
question = AllQuestions(question_id) question = AllQuestions(question_id)
session.add(question) session.add(question)
if data["create_multiple_choice"]: if data["create_multiple_choice"]:
@ -253,88 +278,104 @@ def add_question(data) -> AllQuestions:
return question return question
def get_all_questions() -> List[AllQuestions]: def get_all_questions() -> Sequence[AllQuestions]:
session = get_session() session = get_session()
return session.query(AllQuestions).all() return session.scalars(select(AllQuestions)).all()
def get_all_hidden_answer() -> List[HiddenAnswer]: def get_all_hidden_answer() -> Sequence[HiddenAnswer]:
session = get_session() session = get_session()
return session.query(HiddenAnswer).all() return session.scalars(select(HiddenAnswer)).all()
def get_all_multiple_choice() -> List[MultipleChoice]: def get_all_multiple_choice() -> Sequence[MultipleChoice]:
session = get_session() session = get_session()
return session.query(MultipleChoice).all() return session.scalars(select(MultipleChoice)).all()
def get_category_count(category: Union[Type[MultipleChoice], Type[HiddenAnswer], Type[AllQuestions]]) -> int: def get_category_count(category: type[MultipleChoice] | type[HiddenAnswer] | type[AllQuestions]) -> int:
session = get_session() session = get_session()
return session.query(category).count() return session.scalar(select(func.count(category)))
def get_question(category: Union[Type[MultipleChoice], Type[HiddenAnswer], Type[AllQuestions]], question_id: int) -> \ def get_question[T: (MultipleChoice, HiddenAnswer, AllQuestions)](category: type[T], question_id: int) -> T | None:
Optional[Union[MultipleChoice, HiddenAnswer, AllQuestions]]:
session = get_session() session = get_session()
return session.query(category).filter(category.question_id == question_id).one_or_none() select_stmt = select(category).where(category.question_id == question_id)
return session.scalars(select_stmt).one_or_none()
def get_random_question_of_difficulty(category: Union[Type[MultipleChoice], Type[HiddenAnswer]], def get_random_question_of_difficulty(category: type[MultipleChoice] | type[HiddenAnswer], difficulty: Literal[1, 2, 3]):
difficulty: Literal[1, 2, 3]):
session = get_session() session = get_session()
return session.query(category).filter(category.hidden_answer_difficulty == difficulty).order_by(func.random()).first() select_stmt = select(category).where(category.hidden_answer_difficulty == difficulty).order_by(func.random())
return session.scalars(select_stmt).first()
def get_random_hidden_answer(difficulty: Optional[Literal[1, 2, 3]] = None) -> HiddenAnswer: def get_random_hidden_answer(difficulty: Literal[1, 2, 3] | None = None) -> HiddenAnswer:
session = get_session() session = get_session()
select_stmt = select(HiddenAnswer).order_by(func.random())
if difficulty is not None: if difficulty is not None:
return session.query(HiddenAnswer).filter(HiddenAnswer.hidden_answer_difficulty == difficulty).order_by(func.random()).first() select_stmt = select_stmt.where(HiddenAnswer.hidden_answer_difficulty == difficulty)
return session.query(HiddenAnswer).order_by(func.random()).first() return session.scalars(select_stmt).first()
def get_random_multiple_choice(difficulty: Optional[Literal[1, 2, 3]] = None) -> MultipleChoice: def get_random_multiple_choice(difficulty: Literal[1, 2, 3] | None = None) -> MultipleChoice:
session = get_session() session = get_session()
select_stmt = select(MultipleChoice).order_by(func.random())
if difficulty is not None: if difficulty is not None:
return session.query(MultipleChoice).filter(MultipleChoice.multiple_choice_difficulty == difficulty).order_by( select_stmt = select_stmt.where(MultipleChoice.multiple_choice_difficulty == difficulty)
func.random()).first() return session.scalars(select_stmt).first()
return session.query(MultipleChoice).order_by(func.random()).first()
def check_answer(question_id: int, guess: str) -> Tuple[bool, str]: def check_answer(question_id: int, guess: str) -> tuple[bool, str] | None:
session = get_session() session = get_session()
question: AllQuestions = session.query(AllQuestions).filter(AllQuestions.question_id == question_id).one_or_none() select_stmt = select(AllQuestions).where(AllQuestions.question_id == question_id)
question: AllQuestions = session.scalars(select_stmt).one_or_none()
if question: if question:
answer = question.answer answer = question.answer
return answer == guess, answer return answer == guess, answer
return None
def query_all_questions(offset, limit, query: dict = None, sort=None, order=None) -> Tuple[List[AllQuestions], int]: def query_all_questions(offset, limit, query: dict[str, Any], sort=None, order=None) -> tuple[Sequence[AllQuestions], int]:
session = get_session() session = get_session()
query_params = [] select_stmt = select(AllQuestions)
if query is not None: if query is not None:
for key in query.keys(): for key in query.keys():
if key == "multiple_choice" or key == "hidden_answer": if key == "multiple_choice" or key == "hidden_answer":
if query[key]: if query[key]:
query_params.append(getattr(AllQuestions, key) != None) select_stmt = select_stmt.where(getattr(AllQuestions, key).is_(None))
else: else:
query_params.append(getattr(AllQuestions, key) == None) select_stmt = select_stmt.where(getattr(AllQuestions, key).is_(None))
else: else:
query_params.append(getattr(AllQuestions, key).ilike("%" + query[key] + "%")) select_stmt = select_stmt.where(getattr(AllQuestions, key).ilike("%" + query[key] + "%"))
order_by = None
if sort and order: if sort and order:
order_by = getattr(getattr(AllQuestions, sort), order)() select_stmt = select_stmt.order_by(getattr(getattr(AllQuestions, sort), order)())
q = session.query(AllQuestions).filter(*query_params).order_by(order_by) questions = session.scalars(select_stmt.offset(offset).limit(limit)).all()
questions = q.offset(offset).limit(limit).all() count = session.scalar(func.count(select_stmt))
count = q.count()
return questions, count return questions, count
def get_unhealthy_questions() -> List[AllQuestions]: def get_unhealthy_questions() -> list[AllQuestions]:
session = get_session() session = get_session()
questions = [] questions = []
missing_category_questions = session.query(AllQuestions).filter(and_(AllQuestions.hidden_answer == None, AllQuestions.multiple_choice == None)).all() missing_category_questions = (
multiple_choice_difficulty = session.query(AllQuestions).filter(AllQuestions.multiple_choice.has(MultipleChoice.multiple_choice_difficulty == None)).all() session.scalars(
hidden_answer_difficulty = session.query(AllQuestions).filter(AllQuestions.hidden_answer.has(HiddenAnswer.hidden_answer_difficulty == None)).all() select(AllQuestions)
.where(and_(AllQuestions.hidden_answer.is_(None), AllQuestions.multiple_choice.is_(None))),
).all()
)
multiple_choice_difficulty = (
session.scalars(
select(AllQuestions)
.where(AllQuestions.multiple_choice.has(MultipleChoice.multiple_choice_difficulty.is_(None))),
).all()
)
hidden_answer_difficulty = (
session.scalars(
select(AllQuestions)
.where(AllQuestions.hidden_answer.has(HiddenAnswer.hidden_answer_difficulty.is_(None))),
).all()
)
questions.extend(missing_category_questions) questions.extend(missing_category_questions)
questions.extend(multiple_choice_difficulty) questions.extend(multiple_choice_difficulty)
questions.extend(hidden_answer_difficulty) questions.extend(hidden_answer_difficulty)
@ -343,7 +384,8 @@ def get_unhealthy_questions() -> List[AllQuestions]:
def update_question(question_id, data): def update_question(question_id, data):
session = get_session() session = get_session()
question: AllQuestions = session.query(AllQuestions).get(question_id) session.get(AllQuestions, question_id)
question = session.get(AllQuestions, question_id)
if data["create_multiple_choice"]: if data["create_multiple_choice"]:
multiple_choice = MultipleChoice(question_id) multiple_choice = MultipleChoice(question_id)
session.add(multiple_choice) session.add(multiple_choice)
@ -371,7 +413,7 @@ def add_user(email, password, username=None, **kwargs):
username = email username = email
user = user_datastore.create_user(email=email, password=password_hash, username=username, **kwargs) user = user_datastore.create_user(email=email, password=password_hash, username=username, **kwargs)
session.add(user) session.add(user)
role = session.query(Role).filter(Role.name == "basic").one_or_none() role = session.scalar(select(Role).where(Role.name == "basic"))
if role is not None: if role is not None:
user_datastore.add_role_to_user(user, role) user_datastore.add_role_to_user(user, role)
# user.add_role(role) # user.add_role(role)
@ -387,24 +429,30 @@ def add_user(email, password, username=None, **kwargs):
def get_user(user_id=None, google_id=None) -> User: def get_user(user_id=None, google_id=None) -> User:
session = get_session() session = get_session()
if user_id is not None: if user_id is not None:
user = session.query(User).filter(User.user_id == user_id).one_or_none() user = session.scalars(
select(User).where(User.user_id == user_id),
).one_or_none()
else: else:
user = session.query(User).filter(User.google_id == google_id).one_or_none() user = session.scalars(
select(User).where(User.google_id == google_id),
).one_or_none()
return user return user
def update_user(user_id, data): def update_user(user_id, data):
session = get_session() session = get_session()
user_datastore = get_user_datastore() user_datastore = get_user_datastore()
user = session.query(User).filter(User.user_id == user_id).one_or_none() select_stmt = select(User).where(User.user_id == user_id)
user = session.scalars(select_stmt).one_or_none()
for role in user.roles: for role in user.roles:
if role.name not in data["roles"]: if role.name not in data["roles"]:
# user.remove_role(role) # user.remove_role(role)
user_datastore.remove_role_from_user(user, role) user_datastore.remove_role_from_user(user, role)
for role_name in data["roles"]: for role_name in data["roles"]:
role = session.query(Role).filter(Role.name == role_name).one_or_none() select_stmt = select(Role).where(Role.name == role_name)
role = session.scalars(select_stmt).one_or_none()
# if role is not None: # if role is not None:
# user.add_role(role) # user.add_role(role)
user_datastore.add_role_to_user(user, role) user_datastore.add_role_to_user(user, role)
for column in data.keys(): for column in data.keys():
if column in User.__table__.columns: if column in User.__table__.columns:
@ -412,38 +460,46 @@ def update_user(user_id, data):
session.commit() session.commit()
def query_users(offset, limit, query: dict = None, sort=None, order=None) -> Tuple[List[User], int]: def query_users(offset, limit, query: dict[str, Any], sort=None, order=None) -> tuple[Sequence[User], int]:
session = get_session() session = get_session()
query_params = [] select_stmt = select(User)
if query is not None: if query is not None:
for key in query.keys(): for key in query.keys():
if hasattr(User, key): if hasattr(User, key):
query_params.append(getattr(User, key).ilike("%" + query[key] + "%")) select_stmt = select_stmt.where(getattr(User, key).ilike("%" + query[key] + "%"))
order_by = None
if sort and order: if sort and order:
order_by = getattr(getattr(User, sort), order)() select_stmt = select_stmt.order_by(getattr(getattr(User, sort), order)())
q = session.query(User).filter(*query_params).order_by(order_by) users = session.scalars(select_stmt.offset(offset).limit(limit)).all()
users = q.offset(offset).limit(limit).all() count = session.scalar(func.count(select_stmt))
count = q.count()
return users, count return users, count
def get_all_roles() -> List[Role]: def get_all_roles() -> Sequence[Role]:
session = get_session() session = get_session()
return session.query(Role).all() return session.scalars(select(Role)).all()
def next_hidden_answer(previous_questions: List[int], difficulty) -> HiddenAnswer: def next_hidden_answer(previous_questions: list[int], difficulty) -> HiddenAnswer:
session = get_session() session = get_session()
return session.query(HiddenAnswer)\ select_stmt = (
.filter(HiddenAnswer.hidden_answer_difficulty == difficulty, select(HiddenAnswer)
HiddenAnswer.question_id.notin_(previous_questions))\ .where(
.order_by(func.random()).first() HiddenAnswer.hidden_answer_difficulty == difficulty,
HiddenAnswer.question_id.notin_(previous_questions),
)
.order_by(func.random())
)
return session.scalars(select_stmt).first()
def next_multiple_choice(previous_questions: List[int], difficulty) -> MultipleChoice: def next_multiple_choice(previous_questions: list[int], difficulty) -> MultipleChoice:
session = get_session() session = get_session()
return session.query(MultipleChoice)\ select_stmt = (
.filter(MultipleChoice.multiple_choice_difficulty == difficulty, select(MultipleChoice)
MultipleChoice.question_id.notin_(previous_questions))\ .where(
.order_by(func.random()).first() MultipleChoice.multiple_choice_difficulty == difficulty,
MultipleChoice.question_id.notin_(previous_questions),
)
.order_by(func.random())
)
return session.scalars(select_stmt).first()

View File

@ -57,4 +57,4 @@
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.0-beta3/dist/js/bootstrap.bundle.min.js" integrity="sha384-JEW9xMcG8R+pH31jmWH6WWP0WintQrMb4s7ZOdauHnUtxwoG2vI5DkLtS3qm9Ekf" crossorigin="anonymous"></script> <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.0-beta3/dist/js/bootstrap.bundle.min.js" integrity="sha384-JEW9xMcG8R+pH31jmWH6WWP0WintQrMb4s7ZOdauHnUtxwoG2vI5DkLtS3qm9Ekf" crossorigin="anonymous"></script>
{% block scripts %}{% endblock %} {% block scripts %}{% endblock %}
</body> </body>
</html> </html>

View File

@ -5,4 +5,4 @@
<h1>{{ error_code }}</h1> <h1>{{ error_code }}</h1>
<p>{{ error_msg }}</p> <p>{{ error_msg }}</p>
</div> </div>
{% endblock %} {% endblock %}

View File

@ -37,4 +37,4 @@
</div> </div>
</div> </div>
</div> </div>
{% endblock %} {% endblock %}

View File

@ -1,4 +1,4 @@
runtime: python39 runtime: python39
entrypoint: gunicorn -b :$PORT QuizTheWord.app:app entrypoint: gunicorn -b :$PORT QuizTheWord.app:app
env_variables: env_variables:
CONFIGURATION_SETUP: "QuizTheWord.config.Production" CONFIGURATION_SETUP: "QuizTheWord.config.Production"

5
pyproject.toml Normal file
View File

@ -0,0 +1,5 @@
[tool.autopep8]
max_line_length = 150
[tool.mypy]
check_untyped_defs = true

View File

@ -1,21 +1,13 @@
click~=7.1.2 Authlib~=1.2.1
Flask~=1.1.2 bcrypt~=4.0.1
itsdangerous~=1.1.0 Flask~=3.0.0
Jinja2~=2.11.3 Flask-Security-Too~=5.3.2
MarkupSafe~=1.1.1 gunicorn~=21.2.0
pg8000~=1.17.0
psycopg2~=2.8.6
python-dotenv~=0.15.0
scramp~=1.2.0
SQLAlchemy~=1.3.23
Werkzeug~=1.0.1
Flask-Security-Too~=4.0.0
pytest~=6.2.2
gunicorn~=20.1.0
html5validate~=0.0.2 html5validate~=0.0.2
Authlib~=0.15.4 psycopg[binary]~=3.1.12
Flask-WTF~=0.14.3 pytest~=7.4.3
bcrypt~=3.2.0 python-dotenv~=1.0.0
WTForms~=2.3.3 requests~=2.31.0
html5validate~=0.0.2 setuptools~=68.2.2
requests~=2.26.0 SQLAlchemy~=2.0.23
Werkzeug~=3.0.1

View File

@ -1,3 +1,5 @@
from __future__ import annotations
import pathlib import pathlib
@ -10,7 +12,8 @@ def parse_multiple_choice_questions(question_file: pathlib.Path):
correct_answer_index = int(question_data[0])-1 correct_answer_index = int(question_data[0])-1
question = question_data[1] question = question_data[1]
answer_list = question_data[2].split("//") answer_list = question_data[2].split("//")
bible_verse = question_data[3] if question_data[3].lower() != "none" else None bible_verse = question_data[3] if question_data[3].lower(
) != "none" else None
correct_answer = answer_list.pop(correct_answer_index) correct_answer = answer_list.pop(correct_answer_index)
questions.append({ questions.append({
"question": question, "question": question,

View File

@ -1,6 +1,9 @@
from __future__ import annotations
import json import json
from html5validate import validate
from flask.testing import FlaskClient from flask.testing import FlaskClient
from html5validate import validate
def test_admin_home_with_basic_user(client: FlaskClient, login_basic_user): def test_admin_home_with_basic_user(client: FlaskClient, login_basic_user):

View File

@ -1,7 +1,19 @@
from __future__ import annotations
import os import os
from flask_security import hash_password, SQLAlchemySessionUserDatastore
from QuizTheWord.database import User, Role, AllQuestions, MultipleChoice, HiddenAnswer, get_session, init_db, destroy_db
import pytest import pytest
from flask_security import hash_password
from flask_security import SQLAlchemySessionUserDatastore
from QuizTheWord.database import AllQuestions
from QuizTheWord.database import destroy_db
from QuizTheWord.database import get_session
from QuizTheWord.database import HiddenAnswer
from QuizTheWord.database import init_db
from QuizTheWord.database import MultipleChoice
from QuizTheWord.database import Role
from QuizTheWord.database import User
@pytest.fixture(scope="module") @pytest.fixture(scope="module")