Add typing to general app files

This commit is contained in:
Jan-Pieter Baert 2019-09-08 01:58:21 +02:00
parent b7fa92a22a
commit 15794b06b7
No known key found for this signature in database
GPG key ID: B19186932178234A
8 changed files with 47 additions and 35 deletions

View file

@ -1,12 +1,14 @@
import flask_login as login import flask_login as login
from flask import Flask
from flask_admin import Admin from flask_admin import Admin
from flask_admin.contrib.sqla import ModelView from flask_admin.contrib.sqla import ModelView
from flask_sqlalchemy import SQLAlchemy
from models import Location, Order, OrderItem, Product, User from models import Location, Order, OrderItem, Product, User
class ModelBaseView(ModelView): class ModelBaseView(ModelView):
def is_accessible(self): def is_accessible(self) -> bool:
if login.current_user.is_anonymous(): if login.current_user.is_anonymous():
return False return False
@ -29,7 +31,7 @@ class LocationAdminModel(ModelBaseView):
form_columns = ("name", "address", "website", "telephone") form_columns = ("name", "address", "website", "telephone")
def init_admin(app, db): def init_admin(app: Flask, db: SQLAlchemy) -> None:
admin = Admin(app, name="Haldis", url="/admin", template_mode="bootstrap3") admin = Admin(app, name="Haldis", url="/admin", template_mode="bootstrap3")
admin.add_view(UserAdminModel(User, db.session)) admin.add_view(UserAdminModel(User, db.session))

View file

@ -1,4 +1,5 @@
import logging import logging
import typing
from datetime import datetime from datetime import datetime
from logging.handlers import TimedRotatingFileHandler from logging.handlers import TimedRotatingFileHandler
@ -19,7 +20,7 @@ from utils import euro_string
from zeus import init_oauth from zeus import init_oauth
def create_app(): def create_app() -> Manager:
app = Flask(__name__) app = Flask(__name__)
# Load the config file # Load the config file
@ -34,7 +35,7 @@ def create_app():
return manager return manager
def register_plugins(app, debug: bool): def register_plugins(app: Flask, debug: bool) -> Manager:
# Register Airbrake and enable the logrotation # Register Airbrake and enable the logrotation
if not app.debug: if not app.debug:
timedFileHandler = TimedRotatingFileHandler( timedFileHandler = TimedRotatingFileHandler(
@ -96,17 +97,17 @@ def register_plugins(app, debug: bool):
return manager return manager
def add_handlers(app): def add_handlers(app: Flask) -> None:
@app.errorhandler(404) @app.errorhandler(404)
def handle404(e): def handle404(e) -> typing.Tuple[str, int]:
return render_template("errors/404.html"), 404 return render_template("errors/404.html"), 404
@app.errorhandler(401) @app.errorhandler(401)
def handle401(e): def handle401(e) -> typing.Tuple[str, int]:
return render_template("errors/401.html"), 401 return render_template("errors/401.html"), 401
def add_routes(application): def add_routes(application: Flask) -> None:
# import views # TODO convert to blueprint # import views # TODO convert to blueprint
# import views.stats # TODO convert to blueprint # import views.stats # TODO convert to blueprint
@ -127,9 +128,9 @@ def add_routes(application):
application.register_blueprint(debug_bp, url_prefix="/debug") application.register_blueprint(debug_bp, url_prefix="/debug")
def add_template_filters(app): def add_template_filters(app: Flask) -> None:
@app.template_filter("countdown") @app.template_filter("countdown")
def countdown(value, only_positive=True, show_text=True): def countdown(value, only_positive: bool = True, show_text: bool = True) -> str:
delta = value - datetime.now() delta = value - datetime.now()
if delta.total_seconds() < 0 and only_positive: if delta.total_seconds() < 0 and only_positive:
return "closed" return "closed"
@ -141,11 +142,11 @@ def add_template_filters(app):
return time return time
@app.template_filter("year") @app.template_filter("year")
def current_year(value): def current_year(value: typing.Any) -> str:
return str(datetime.now().year) return str(datetime.now().year)
@app.template_filter("euro") @app.template_filter("euro")
def euro(value): def euro(value: int) -> None:
euro_string(value) euro_string(value)

View file

@ -1,3 +1,5 @@
import typing
from sqlalchemy.sql import desc, func from sqlalchemy.sql import desc, func
from models import Location, Order, OrderItem, Product, User from models import Location, Order, OrderItem, Product, User
@ -42,7 +44,7 @@ class FatOrderItem(OrderItem, FatModel):
class FatProduct(Product, FatModel): class FatProduct(Product, FatModel):
@classmethod @classmethod
def top4(cls): def top4(cls) -> None:
top4 = ( top4 = (
OrderItem.query.join(Product) OrderItem.query.join(Product)
.join(Location) .join(Location)

View file

@ -3,7 +3,8 @@ from datetime import datetime, timedelta
from flask import session from flask import session
from flask_login import current_user from flask_login import current_user
from flask_wtf import FlaskForm as Form from flask_wtf import FlaskForm as Form
from wtforms import DateTimeField, SelectField, StringField, SubmitField, validators from wtforms import (DateTimeField, SelectField, StringField, SubmitField,
validators)
from models import Location, User from models import Location, User
from utils import euro_string from utils import euro_string
@ -20,7 +21,7 @@ class OrderForm(Form):
stoptime = DateTimeField("Stoptime", format="%d-%m-%Y %H:%M") stoptime = DateTimeField("Stoptime", format="%d-%m-%Y %H:%M")
submit_button = SubmitField("Submit") submit_button = SubmitField("Submit")
def populate(self): def populate(self) -> None:
if current_user.is_admin(): if current_user.is_admin():
self.courrier_id.choices = [(0, None)] + [ self.courrier_id.choices = [(0, None)] + [
(u.id, u.username) for u in User.query.order_by("username") (u.id, u.username) for u in User.query.order_by("username")
@ -42,7 +43,7 @@ class OrderItemForm(Form):
extra = StringField("Extra") extra = StringField("Extra")
submit_button = SubmitField("Submit") submit_button = SubmitField("Submit")
def populate(self, location): def populate(self, location: Location) -> None:
self.product_id.choices = [ self.product_id.choices = [
(i.id, (i.name + ": " + euro_string(i.price))) for i in location.products (i.id, (i.name + ": " + euro_string(i.price))) for i in location.products
] ]
@ -51,12 +52,12 @@ class OrderItemForm(Form):
class AnonOrderItemForm(OrderItemForm): class AnonOrderItemForm(OrderItemForm):
name = StringField("Name", validators=[validators.required()]) name = StringField("Name", validators=[validators.required()])
def populate(self, location): def populate(self, location: Location) -> None:
OrderItemForm.populate(self, location) OrderItemForm.populate(self, location)
if self.name.data is None: if self.name.data is None:
self.name.data = session.get("anon_name", None) self.name.data = session.get("anon_name", None)
def validate(self): def validate(self) -> bool:
rv = OrderForm.validate(self) rv = OrderForm.validate(self)
if not rv: if not rv:
return False return False

View file

@ -1,6 +1,6 @@
from flask import abort, Blueprint from flask import Blueprint, abort, redirect, session, url_for
from flask import redirect, session, url_for
from flask_login import current_user, logout_user from flask_login import current_user, logout_user
from werkzeug.wrappers import Response
from models import User from models import User
from zeus import zeus_login from zeus import zeus_login
@ -8,9 +8,9 @@ from zeus import zeus_login
auth_bp = Blueprint("auth_bp", __name__) auth_bp = Blueprint("auth_bp", __name__)
def init_login(app): def init_login(app) -> None:
@app.login_manager.user_loader @app.login_manager.user_loader
def load_user(userid): def load_user(userid) -> User:
return User.query.filter_by(id=userid).first() return User.query.filter_by(id=userid).first()
@ -20,13 +20,13 @@ def login():
@auth_bp.route("/logout") @auth_bp.route("/logout")
def logout(): def logout() -> Response:
if "zeus_token" in session: if "zeus_token" in session:
session.pop("zeus_token", None) session.pop("zeus_token", None)
logout_user() logout_user()
return redirect(url_for("general_bp.home")) return redirect(url_for("general_bp.home"))
def before_request(): def before_request() -> None:
if current_user.is_anonymous() or not current_user.is_allowed(): if current_user.is_anonymous() or not current_user.is_allowed():
abort(401) abort(401)

View file

@ -7,7 +7,7 @@ from flask import current_app as app
from flask import url_for from flask import url_for
def post_order_to_webhook(order_item): def post_order_to_webhook(order_item) -> None:
message = "" message = ""
if order_item.courrier is not None: if order_item.courrier is not None:
message = "<!channel|@channel> {3} is going to {1}, order <{0}|here>! Deadline in {2} minutes!".format( message = "<!channel|@channel> {3} is going to {1}, order <{0}|here>! Deadline in {2} minutes!".format(
@ -27,14 +27,14 @@ def post_order_to_webhook(order_item):
class WebhookSenderThread(Thread): class WebhookSenderThread(Thread):
def __init__(self, message): def __init__(self, message: str) -> None:
super(WebhookSenderThread, self).__init__() super(WebhookSenderThread, self).__init__()
self.message = message self.message = message
def run(self): def run(self) -> None:
self.slack_webhook() self.slack_webhook()
def slack_webhook(self): def slack_webhook(self) -> None:
js = json.dumps({"text": self.message}) js = json.dumps({"text": self.message})
url = app.config["SLACK_WEBHOOK"] url = app.config["SLACK_WEBHOOK"]
if len(url) > 0: if len(url) > 0:
@ -43,7 +43,7 @@ class WebhookSenderThread(Thread):
app.logger.info(str(js)) app.logger.info(str(js))
def remaining_minutes(value): def remaining_minutes(value) -> str:
delta = value - datetime.now() delta = value - datetime.now()
if delta.total_seconds() < 0: if delta.total_seconds() < 0:
return "0" return "0"

View file

@ -1,4 +1,4 @@
def euro_string(value): def euro_string(value: int) -> str:
""" """
Convert cents to string formatted euro Convert cents to string formatted euro
""" """

View file

@ -1,6 +1,10 @@
from flask import current_app, flash, redirect, request, session, url_for, Blueprint import typing
from flask import (Blueprint, current_app, flash, redirect, request, session,
url_for)
from flask_login import login_user from flask_login import login_user
from flask_oauthlib.client import OAuthException, OAuth from flask_oauthlib.client import OAuth, OAuthException
from werkzeug.wrappers import Response
from models import User, db from models import User, db
@ -14,7 +18,9 @@ def zeus_login():
@oauth_bp.route("/login/zeus/authorized") @oauth_bp.route("/login/zeus/authorized")
def authorized(): def authorized() -> typing.Any:
# type is 'typing.Union[str, Response]', but this errors due to
# https://github.com/python/mypy/issues/7187
resp = current_app.zeus.authorized_response() resp = current_app.zeus.authorized_response()
if resp is None: if resp is None:
return "Access denied: reason=%s error=%s" % ( return "Access denied: reason=%s error=%s" % (
@ -60,12 +66,12 @@ def init_oauth(app):
return zeus return zeus
def login_and_redirect_user(user): def login_and_redirect_user(user) -> Response:
login_user(user) login_user(user)
return redirect(url_for("general_bp.home")) return redirect(url_for("general_bp.home"))
def create_user(username): def create_user(username) -> User:
user = User() user = User()
user.configure(username, False, 1) user.configure(username, False, 1)
db.session.add(user) db.session.add(user)