commit
32c60eaff5
28 changed files with 178 additions and 141 deletions
|
@ -1,12 +1,14 @@
|
|||
import flask_login as login
|
||||
from flask import Flask
|
||||
from flask_admin import Admin
|
||||
from flask_admin.contrib.sqla import ModelView
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
|
||||
from models import Location, Order, OrderItem, Product, User
|
||||
|
||||
|
||||
class ModelBaseView(ModelView):
|
||||
def is_accessible(self):
|
||||
def is_accessible(self) -> bool:
|
||||
if login.current_user.is_anonymous():
|
||||
return False
|
||||
|
||||
|
@ -29,7 +31,7 @@ class LocationAdminModel(ModelBaseView):
|
|||
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.add_view(UserAdminModel(User, db.session))
|
||||
|
|
21
app/app.py
21
app/app.py
|
@ -1,4 +1,5 @@
|
|||
import logging
|
||||
import typing
|
||||
from datetime import datetime
|
||||
from logging.handlers import TimedRotatingFileHandler
|
||||
|
||||
|
@ -19,7 +20,7 @@ from utils import euro_string
|
|||
from zeus import init_oauth
|
||||
|
||||
|
||||
def create_app():
|
||||
def create_app() -> Manager:
|
||||
app = Flask(__name__)
|
||||
|
||||
# Load the config file
|
||||
|
@ -34,7 +35,7 @@ def create_app():
|
|||
return manager
|
||||
|
||||
|
||||
def register_plugins(app, debug: bool):
|
||||
def register_plugins(app: Flask, debug: bool) -> Manager:
|
||||
# Register Airbrake and enable the logrotation
|
||||
if not app.debug:
|
||||
timedFileHandler = TimedRotatingFileHandler(
|
||||
|
@ -96,17 +97,17 @@ def register_plugins(app, debug: bool):
|
|||
return manager
|
||||
|
||||
|
||||
def add_handlers(app):
|
||||
def add_handlers(app: Flask) -> None:
|
||||
@app.errorhandler(404)
|
||||
def handle404(e):
|
||||
def handle404(e) -> typing.Tuple[str, int]:
|
||||
return render_template("errors/404.html"), 404
|
||||
|
||||
@app.errorhandler(401)
|
||||
def handle401(e):
|
||||
def handle401(e) -> typing.Tuple[str, int]:
|
||||
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.stats # TODO convert to blueprint
|
||||
|
||||
|
@ -127,9 +128,9 @@ def add_routes(application):
|
|||
application.register_blueprint(debug_bp, url_prefix="/debug")
|
||||
|
||||
|
||||
def add_template_filters(app):
|
||||
def add_template_filters(app: Flask) -> None:
|
||||
@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()
|
||||
if delta.total_seconds() < 0 and only_positive:
|
||||
return "closed"
|
||||
|
@ -141,11 +142,11 @@ def add_template_filters(app):
|
|||
return time
|
||||
|
||||
@app.template_filter("year")
|
||||
def current_year(value):
|
||||
def current_year(value: typing.Any) -> str:
|
||||
return str(datetime.now().year)
|
||||
|
||||
@app.template_filter("euro")
|
||||
def euro(value):
|
||||
def euro(value: int) -> None:
|
||||
euro_string(value)
|
||||
|
||||
|
||||
|
|
|
@ -2,7 +2,7 @@ from app import db
|
|||
from models import User
|
||||
|
||||
|
||||
def add():
|
||||
def add() -> None:
|
||||
feli = User()
|
||||
feli.configure("feliciaan", True, 0)
|
||||
db.session.add(feli)
|
||||
|
|
|
@ -1,6 +1,5 @@
|
|||
from models import Location, Product
|
||||
from app import db
|
||||
|
||||
from models import Location, Product
|
||||
|
||||
menuitems = [
|
||||
"Spicy Chicken",
|
||||
|
@ -23,7 +22,7 @@ menuitems = [
|
|||
pricedict = {"Small": 799, "Medium": 999, "Large": 1199}
|
||||
|
||||
|
||||
def add():
|
||||
def add() -> None:
|
||||
simpizza = Location()
|
||||
simpizza.configure("Fitchen", "?", "?", "https://www.fitchen.be/")
|
||||
db.session.add(simpizza)
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
from models import Location, Product
|
||||
from app import db
|
||||
from itertools import product
|
||||
|
||||
from app import db
|
||||
from models import Location, Product
|
||||
|
||||
zetmelen = ["Nasi", "Bami"]
|
||||
vlezen = ["Rundsvlees", "Varkensvlees", "Kippenstukkjes"]
|
||||
|
@ -29,7 +29,7 @@ specials = [
|
|||
]
|
||||
|
||||
|
||||
def add():
|
||||
def add() -> None:
|
||||
chinees = Location()
|
||||
chinees.configure(
|
||||
"Oceans's Garden",
|
||||
|
@ -39,12 +39,12 @@ def add():
|
|||
)
|
||||
db.session.add(chinees)
|
||||
|
||||
def chinees_create_entry(name):
|
||||
def chinees_create_entry(name) -> None:
|
||||
entry = Product()
|
||||
entry.configure(chinees, name, 550)
|
||||
db.session.add(entry)
|
||||
|
||||
def chinees_create_regulat(zetmeel, vlees="", saus=""):
|
||||
def chinees_create_regulat(zetmeel, vlees="", saus="") -> None:
|
||||
chinees_create_entry("{} {} {}".format(zetmeel, vlees, saus).rstrip())
|
||||
|
||||
for z, v, s in product(zetmelen, vlezen, sauzen):
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
from models import Location, Product
|
||||
from app import db
|
||||
from models import Location, Product
|
||||
|
||||
|
||||
def add():
|
||||
|
@ -46,7 +46,7 @@ pizzasTA = {
|
|||
}
|
||||
|
||||
|
||||
def addTA():
|
||||
def addTA() -> None:
|
||||
primadonna_takeaway = Location()
|
||||
primadonna_takeaway.configure(
|
||||
"Primadonna (takeaway laten bezorgen)",
|
||||
|
@ -101,7 +101,7 @@ pizzasAfhalen = {
|
|||
}
|
||||
|
||||
|
||||
def addAfhalen():
|
||||
def addAfhalen() -> None:
|
||||
primadonna_afhalen = Location()
|
||||
primadonna_afhalen.configure(
|
||||
"Primadonna (bellen en afhalen)",
|
||||
|
|
|
@ -1,6 +1,5 @@
|
|||
from models import Location, Product
|
||||
from app import db
|
||||
|
||||
from models import Location, Product
|
||||
|
||||
pizzas = [
|
||||
"Bolognese de luxe",
|
||||
|
@ -33,7 +32,7 @@ pizzas = [
|
|||
]
|
||||
|
||||
|
||||
def add():
|
||||
def add() -> None:
|
||||
simpizza = Location()
|
||||
simpizza.configure(
|
||||
"Sim-pizza",
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
from models import Location, Product
|
||||
from app import db
|
||||
from models import Location, Product
|
||||
|
||||
bickies = {
|
||||
"Bicky Burger Original": 330,
|
||||
|
@ -11,7 +11,7 @@ bickies = {
|
|||
"Bicky Veggie": 350,
|
||||
}
|
||||
|
||||
saus = {
|
||||
sauskes = {
|
||||
"american": 70,
|
||||
"andalouse": 70,
|
||||
"bicky saus": 70,
|
||||
|
@ -100,7 +100,7 @@ friet = {"Klein pak": 200, "Midden pak": 250, "Groot pak": 300}
|
|||
data = [special_bickies, specials, vlezekes, friet]
|
||||
|
||||
|
||||
def add():
|
||||
def add() -> None:
|
||||
stefanos = Location()
|
||||
stefanos.configure(
|
||||
"Stefano's Place",
|
||||
|
@ -127,7 +127,7 @@ def add():
|
|||
db.session.add(item)
|
||||
|
||||
# saus in een potteke bestellen is 10 cent extra
|
||||
for name, price in saus.items():
|
||||
for name, price in sauskes.items():
|
||||
saus = Product()
|
||||
saus.configure(stefanos, name, price)
|
||||
db.session.add(saus)
|
||||
|
|
|
@ -1,6 +1,9 @@
|
|||
import add_admins
|
||||
import add_fitchen
|
||||
import add_oceans_garden
|
||||
import add_primadonna
|
||||
import add_simpizza
|
||||
from app import db
|
||||
import add_oceans_garden, add_admins, add_simpizza, add_primadonna, add_fitchen
|
||||
|
||||
|
||||
entry_sets = {
|
||||
"Admins": add_admins.add,
|
||||
|
@ -15,23 +18,23 @@ no = ["no", "n", "N"]
|
|||
|
||||
|
||||
# Commit all the things
|
||||
def commit():
|
||||
def commit() -> None:
|
||||
db.session.commit()
|
||||
print("Committing successful")
|
||||
|
||||
|
||||
def check_if_overwrite():
|
||||
def check_if_overwrite() -> bool:
|
||||
answer = input("Do you want to overwrite the previous database? (y/N) ")
|
||||
return answer in yes
|
||||
|
||||
|
||||
def add_all():
|
||||
def add_all() -> None:
|
||||
for entry_set in entry_sets.keys():
|
||||
print("Adding {}.".format(entry_set))
|
||||
entry_sets[entry_set]()
|
||||
|
||||
|
||||
def recreate_from_scratch():
|
||||
def recreate_from_scratch() -> None:
|
||||
confirmation = "Are you very very sure? (Will delete previous entries!) (y/N) "
|
||||
check = "I acknowledge any repercussions!"
|
||||
if input(confirmation) in yes and input("Type: '{}' ".format(check)) == check:
|
||||
|
@ -41,10 +44,10 @@ def recreate_from_scratch():
|
|||
add_to_current()
|
||||
|
||||
|
||||
def add_to_current():
|
||||
def add_to_current() -> None:
|
||||
available = [entry_set for entry_set in entry_sets]
|
||||
|
||||
def add_numbers():
|
||||
def add_numbers() -> str:
|
||||
return " ".join(
|
||||
["{}({}), ".format(loc, i) for i, loc in enumerate(available)]
|
||||
).rstrip(", ")
|
||||
|
@ -59,17 +62,17 @@ def add_to_current():
|
|||
available = []
|
||||
elif answer == "C":
|
||||
pass
|
||||
elif answer in [str(x) for x in range(len(available))]:
|
||||
answer = int(answer)
|
||||
print("Adding {}.".format(available[answer]))
|
||||
entry_sets[str(available[answer])]()
|
||||
del available[answer]
|
||||
elif answer.isnumeric() and answer in [str(x) for x in range(len(available))]:
|
||||
answer_index = int(answer)
|
||||
print("Adding {}.".format(available[answer_index]))
|
||||
entry_sets[str(available[answer_index])]()
|
||||
del available[answer_index]
|
||||
else:
|
||||
print("Not a valid answer.")
|
||||
print("Thank you for adding, come again!")
|
||||
|
||||
|
||||
def init():
|
||||
def init() -> None:
|
||||
print("Database modification script!")
|
||||
print("=============================\n\n")
|
||||
if check_if_overwrite():
|
||||
|
|
|
@ -1,3 +1,5 @@
|
|||
import typing
|
||||
|
||||
from sqlalchemy.sql import desc, func
|
||||
|
||||
from models import Location, Order, OrderItem, Product, User
|
||||
|
@ -42,7 +44,7 @@ class FatOrderItem(OrderItem, FatModel):
|
|||
|
||||
class FatProduct(Product, FatModel):
|
||||
@classmethod
|
||||
def top4(cls):
|
||||
def top4(cls) -> None:
|
||||
top4 = (
|
||||
OrderItem.query.join(Product)
|
||||
.join(Location)
|
||||
|
|
11
app/forms.py
11
app/forms.py
|
@ -3,7 +3,8 @@ from datetime import datetime, timedelta
|
|||
from flask import session
|
||||
from flask_login import current_user
|
||||
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 utils import euro_string
|
||||
|
@ -20,7 +21,7 @@ class OrderForm(Form):
|
|||
stoptime = DateTimeField("Stoptime", format="%d-%m-%Y %H:%M")
|
||||
submit_button = SubmitField("Submit")
|
||||
|
||||
def populate(self):
|
||||
def populate(self) -> None:
|
||||
if current_user.is_admin():
|
||||
self.courrier_id.choices = [(0, None)] + [
|
||||
(u.id, u.username) for u in User.query.order_by("username")
|
||||
|
@ -42,7 +43,7 @@ class OrderItemForm(Form):
|
|||
extra = StringField("Extra")
|
||||
submit_button = SubmitField("Submit")
|
||||
|
||||
def populate(self, location):
|
||||
def populate(self, location: Location) -> None:
|
||||
self.product_id.choices = [
|
||||
(i.id, (i.name + ": " + euro_string(i.price))) for i in location.products
|
||||
]
|
||||
|
@ -51,12 +52,12 @@ class OrderItemForm(Form):
|
|||
class AnonOrderItemForm(OrderItemForm):
|
||||
name = StringField("Name", validators=[validators.required()])
|
||||
|
||||
def populate(self, location):
|
||||
def populate(self, location: Location) -> None:
|
||||
OrderItemForm.populate(self, location)
|
||||
if self.name.data is None:
|
||||
self.name.data = session.get("anon_name", None)
|
||||
|
||||
def validate(self):
|
||||
def validate(self) -> bool:
|
||||
rv = OrderForm.validate(self)
|
||||
if not rv:
|
||||
return False
|
||||
|
|
12
app/login.py
12
app/login.py
|
@ -1,6 +1,6 @@
|
|||
from flask import abort, Blueprint
|
||||
from flask import redirect, session, url_for
|
||||
from flask import Blueprint, abort, redirect, session, url_for
|
||||
from flask_login import current_user, logout_user
|
||||
from werkzeug.wrappers import Response
|
||||
|
||||
from models import User
|
||||
from zeus import zeus_login
|
||||
|
@ -8,9 +8,9 @@ from zeus import zeus_login
|
|||
auth_bp = Blueprint("auth_bp", __name__)
|
||||
|
||||
|
||||
def init_login(app):
|
||||
def init_login(app) -> None:
|
||||
@app.login_manager.user_loader
|
||||
def load_user(userid):
|
||||
def load_user(userid) -> User:
|
||||
return User.query.filter_by(id=userid).first()
|
||||
|
||||
|
||||
|
@ -20,13 +20,13 @@ def login():
|
|||
|
||||
|
||||
@auth_bp.route("/logout")
|
||||
def logout():
|
||||
def logout() -> Response:
|
||||
if "zeus_token" in session:
|
||||
session.pop("zeus_token", None)
|
||||
logout_user()
|
||||
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():
|
||||
abort(401)
|
||||
|
|
|
@ -1,8 +1,15 @@
|
|||
from __future__ import with_statement
|
||||
from alembic import context
|
||||
from sqlalchemy import engine_from_config, pool
|
||||
|
||||
from logging.config import fileConfig
|
||||
|
||||
from alembic import context
|
||||
# add your model's MetaData object here
|
||||
# for 'autogenerate' support
|
||||
# from myapp import mymodel
|
||||
# target_metadata = mymodel.Base.metadata
|
||||
from flask import current_app
|
||||
from sqlalchemy import engine_from_config, pool
|
||||
|
||||
# this is the Alembic Config object, which provides
|
||||
# access to the values within the .ini file in use.
|
||||
config = context.config
|
||||
|
@ -11,11 +18,6 @@ config = context.config
|
|||
# This line sets up loggers basically.
|
||||
fileConfig(config.config_file_name)
|
||||
|
||||
# add your model's MetaData object here
|
||||
# for 'autogenerate' support
|
||||
# from myapp import mymodel
|
||||
# target_metadata = mymodel.Base.metadata
|
||||
from flask import current_app
|
||||
|
||||
config.set_main_option(
|
||||
"sqlalchemy.url", current_app.config.get("SQLALCHEMY_DATABASE_URI")
|
||||
|
|
|
@ -10,8 +10,8 @@ Create Date: 2019-04-02 18:00:12.618368
|
|||
revision = "150252c1cdb1"
|
||||
down_revision = None
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
|
||||
def upgrade():
|
||||
|
|
|
@ -1,17 +1,17 @@
|
|||
class AnonymouseUser:
|
||||
id = None
|
||||
|
||||
def is_active(self):
|
||||
def is_active(self) -> bool:
|
||||
return False
|
||||
|
||||
def is_authenticated(self):
|
||||
def is_authenticated(self) -> bool:
|
||||
return False
|
||||
|
||||
def is_anonymous(self):
|
||||
def is_anonymous(self) -> bool:
|
||||
return True
|
||||
|
||||
def is_admin(self):
|
||||
def is_admin(self) -> bool:
|
||||
return False
|
||||
|
||||
def get_id(self):
|
||||
def get_id(self) -> None:
|
||||
return None
|
||||
|
|
|
@ -1,3 +1,5 @@
|
|||
import typing
|
||||
|
||||
from models import db
|
||||
|
||||
|
||||
|
@ -10,11 +12,13 @@ class Location(db.Model):
|
|||
products = db.relationship("Product", backref="location", lazy="dynamic")
|
||||
orders = db.relationship("Order", backref="location", lazy="dynamic")
|
||||
|
||||
def configure(self, name, address, telephone, website):
|
||||
def configure(
|
||||
self, name: str, address: str, telephone: typing.Optional[str], website: str
|
||||
) -> None:
|
||||
self.name = name
|
||||
self.address = address
|
||||
self.website = website
|
||||
self.telephone = telephone
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return "%s" % (self.name)
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
import typing
|
||||
from datetime import datetime
|
||||
|
||||
from .database import db
|
||||
from .location import Location
|
||||
from .user import User
|
||||
|
||||
|
||||
|
@ -13,20 +15,26 @@ class Order(db.Model):
|
|||
public = db.Column(db.Boolean, default=True)
|
||||
items = db.relationship("OrderItem", backref="order", lazy="dynamic")
|
||||
|
||||
def configure(self, courrier, location, starttime, stoptime):
|
||||
def configure(
|
||||
self,
|
||||
courrier: User,
|
||||
location: Location,
|
||||
starttime: db.DateTime,
|
||||
stoptime: db.DateTime,
|
||||
) -> None:
|
||||
self.courrier = courrier
|
||||
self.location = location
|
||||
self.starttime = starttime
|
||||
self.stoptime = stoptime
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
if self.location:
|
||||
return "Order %d @ %s" % (self.id, self.location.name or "None")
|
||||
else:
|
||||
return "Order %d" % (self.id)
|
||||
|
||||
def group_by_user(self):
|
||||
group = dict()
|
||||
def group_by_user(self) -> typing.Dict[str, typing.Any]:
|
||||
group: typing.Dict[str, typing.Any] = dict()
|
||||
for item in self.items:
|
||||
user = group.get(item.get_name(), dict())
|
||||
user["total"] = user.get("total", 0) + item.product.price
|
||||
|
@ -39,8 +47,8 @@ class Order(db.Model):
|
|||
|
||||
return group
|
||||
|
||||
def group_by_product(self):
|
||||
group = dict()
|
||||
def group_by_product(self) -> typing.Dict[str, typing.Any]:
|
||||
group: typing.Dict[str, typing.Any] = dict()
|
||||
for item in self.items:
|
||||
product = group.get(item.product.name, dict())
|
||||
product["count"] = product.get("count", 0) + 1
|
||||
|
@ -50,7 +58,7 @@ class Order(db.Model):
|
|||
|
||||
return group
|
||||
|
||||
def can_close(self, user_id):
|
||||
def can_close(self, user_id: int) -> bool:
|
||||
if self.stoptime and self.stoptime < datetime.now():
|
||||
return False
|
||||
user = None
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
from datetime import datetime
|
||||
|
||||
from .database import db
|
||||
from .order import Order
|
||||
from .product import Product
|
||||
from .user import User
|
||||
|
||||
|
||||
|
@ -17,17 +19,17 @@ class OrderItem(db.Model):
|
|||
extra = db.Column(db.String(254), nullable=True)
|
||||
name = db.Column(db.String(120))
|
||||
|
||||
def configure(self, user, order, product):
|
||||
def configure(self, user: User, order: Order, product: Product) -> None:
|
||||
self.user = user
|
||||
self.order = order
|
||||
self.product = product
|
||||
|
||||
def get_name(self):
|
||||
def get_name(self) -> str:
|
||||
if self.user_id is not None and self.user_id > 0:
|
||||
return self.user.username
|
||||
return self.name
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
product_name = None
|
||||
if self.product:
|
||||
product_name = self.product.name
|
||||
|
@ -37,7 +39,7 @@ class OrderItem(db.Model):
|
|||
product_name or "None",
|
||||
)
|
||||
|
||||
def can_delete(self, order_id, user_id, name):
|
||||
def can_delete(self, order_id: int, user_id: int, name: str) -> bool:
|
||||
if int(self.order_id) != int(order_id):
|
||||
return False
|
||||
if self.order.stoptime and self.order.stoptime < datetime.now():
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
from models import db
|
||||
|
||||
from .location import Location
|
||||
|
||||
|
||||
class Product(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
|
@ -8,12 +10,12 @@ class Product(db.Model):
|
|||
price = db.Column(db.Integer, nullable=False)
|
||||
orderItems = db.relationship("OrderItem", backref="product", lazy="dynamic")
|
||||
|
||||
def configure(self, location, name, price):
|
||||
def configure(self, location: Location, name: str, price: int) -> None:
|
||||
self.location = location
|
||||
self.name = name
|
||||
self.price = price
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return "%s (€%d)from %s" % (
|
||||
self.name,
|
||||
self.price / 100,
|
||||
|
|
|
@ -14,25 +14,25 @@ class User(db.Model):
|
|||
)
|
||||
orderItems = db.relationship("OrderItem", backref="user", lazy="dynamic")
|
||||
|
||||
def configure(self, username, admin, bias):
|
||||
def configure(self, username: str, admin: bool, bias: int) -> None:
|
||||
self.username = username
|
||||
self.admin = admin
|
||||
self.bias = bias
|
||||
|
||||
def is_authenticated(self):
|
||||
def is_authenticated(self) -> bool:
|
||||
return True
|
||||
|
||||
def is_active(self):
|
||||
def is_active(self) -> bool:
|
||||
return True
|
||||
|
||||
def is_admin(self):
|
||||
def is_admin(self) -> bool:
|
||||
return self.admin
|
||||
|
||||
def is_anonymous(self):
|
||||
def is_anonymous(self) -> bool:
|
||||
return False
|
||||
|
||||
def get_id(self):
|
||||
def get_id(self) -> str:
|
||||
return str(self.id)
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return "%s" % self.username
|
||||
|
|
|
@ -7,7 +7,7 @@ from flask import current_app as app
|
|||
from flask import url_for
|
||||
|
||||
|
||||
def post_order_to_webhook(order_item):
|
||||
def post_order_to_webhook(order_item) -> None:
|
||||
message = ""
|
||||
if order_item.courrier is not None:
|
||||
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):
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str) -> None:
|
||||
super(WebhookSenderThread, self).__init__()
|
||||
self.message = message
|
||||
|
||||
def run(self):
|
||||
def run(self) -> None:
|
||||
self.slack_webhook()
|
||||
|
||||
def slack_webhook(self):
|
||||
def slack_webhook(self) -> None:
|
||||
js = json.dumps({"text": self.message})
|
||||
url = app.config["SLACK_WEBHOOK"]
|
||||
if len(url) > 0:
|
||||
|
@ -43,7 +43,7 @@ class WebhookSenderThread(Thread):
|
|||
app.logger.info(str(js))
|
||||
|
||||
|
||||
def remaining_minutes(value):
|
||||
def remaining_minutes(value) -> str:
|
||||
delta = value - datetime.now()
|
||||
if delta.total_seconds() < 0:
|
||||
return "0"
|
||||
|
|
|
@ -1,4 +1,4 @@
|
|||
def euro_string(value):
|
||||
def euro_string(value: int) -> str:
|
||||
"""
|
||||
Convert cents to string formatted euro
|
||||
"""
|
||||
|
|
|
@ -8,7 +8,7 @@ debug_bp = Blueprint("debug_bp", __name__)
|
|||
|
||||
@debug_bp.route("/routes")
|
||||
@login_required
|
||||
def list_routes():
|
||||
def list_routes() -> str:
|
||||
import urllib
|
||||
|
||||
output = []
|
||||
|
|
|
@ -7,7 +7,6 @@ from flask import render_template, send_from_directory, url_for
|
|||
from flask_login import login_required
|
||||
|
||||
from models import Location, Order
|
||||
|
||||
# import views
|
||||
from views.order import get_orders
|
||||
|
||||
|
@ -15,7 +14,7 @@ general_bp = Blueprint("general_bp", __name__)
|
|||
|
||||
|
||||
@general_bp.route("/")
|
||||
def home():
|
||||
def home() -> str:
|
||||
prev_day = datetime.now() - timedelta(days=1)
|
||||
recently_closed = get_orders(
|
||||
((Order.stoptime > prev_day) & (Order.stoptime < datetime.now()))
|
||||
|
@ -27,19 +26,19 @@ def home():
|
|||
|
||||
@general_bp.route("/map", defaults={"id": None})
|
||||
@general_bp.route("/map/<int:id>")
|
||||
def map(id):
|
||||
def map(id) -> str:
|
||||
locs = Location.query.order_by("name")
|
||||
return render_template("maps.html", locations=locs)
|
||||
|
||||
|
||||
@general_bp.route("/location")
|
||||
def locations():
|
||||
def locations() -> str:
|
||||
locs = Location.query.order_by("name")
|
||||
return render_template("locations.html", locations=locs)
|
||||
|
||||
|
||||
@general_bp.route("/location/<int:id>")
|
||||
def location(id):
|
||||
def location(id) -> str:
|
||||
loc = Location.query.filter(Location.id == id).first()
|
||||
if loc is None:
|
||||
abort(404)
|
||||
|
@ -47,27 +46,27 @@ def location(id):
|
|||
|
||||
|
||||
@general_bp.route("/about/")
|
||||
def about():
|
||||
def about() -> str:
|
||||
return render_template("about.html")
|
||||
|
||||
|
||||
@general_bp.route("/profile/")
|
||||
@login_required
|
||||
def profile():
|
||||
def profile() -> str:
|
||||
return render_template("profile.html")
|
||||
|
||||
|
||||
@general_bp.route("/favicon.ico")
|
||||
def favicon():
|
||||
def favicon() -> str:
|
||||
if len(get_orders((Order.stoptime > datetime.now()))) == 0:
|
||||
return send_from_directory(
|
||||
os.path.join(app.root_path, "static"),
|
||||
os.path.join(str(app.root_path), "static"),
|
||||
"favicon.ico",
|
||||
mimetype="image/x-icon",
|
||||
)
|
||||
else:
|
||||
return send_from_directory(
|
||||
os.path.join(app.root_path, "static"),
|
||||
os.path.join(str(app.root_path), "static"),
|
||||
"favicon_orange.ico",
|
||||
mimetype="image/x-icon",
|
||||
)
|
||||
|
|
|
@ -1,18 +1,13 @@
|
|||
import random
|
||||
import typing
|
||||
from datetime import datetime
|
||||
|
||||
import werkzeug
|
||||
# from flask import current_app as app
|
||||
from flask import (
|
||||
Blueprint,
|
||||
abort,
|
||||
flash,
|
||||
redirect,
|
||||
render_template,
|
||||
request,
|
||||
session,
|
||||
url_for,
|
||||
)
|
||||
from flask import (Blueprint, abort, flash, redirect, render_template, request,
|
||||
session, url_for, wrappers)
|
||||
from flask_login import current_user, login_required
|
||||
from werkzeug.wrappers import Response
|
||||
|
||||
from forms import AnonOrderItemForm, OrderForm, OrderItemForm
|
||||
from models import Order, OrderItem, User, db
|
||||
|
@ -22,7 +17,7 @@ order_bp = Blueprint("order_bp", "order")
|
|||
|
||||
|
||||
@order_bp.route("/")
|
||||
def orders(form=None):
|
||||
def orders(form: OrderForm = None) -> str:
|
||||
if form is None and not current_user.is_anonymous():
|
||||
form = OrderForm()
|
||||
location_id = request.args.get("location_id")
|
||||
|
@ -34,7 +29,7 @@ def orders(form=None):
|
|||
|
||||
@order_bp.route("/create", methods=["POST"])
|
||||
@login_required
|
||||
def order_create():
|
||||
def order_create() -> typing.Union[str, Response]:
|
||||
orderForm = OrderForm()
|
||||
orderForm.populate()
|
||||
if orderForm.validate_on_submit():
|
||||
|
@ -48,7 +43,7 @@ def order_create():
|
|||
|
||||
|
||||
@order_bp.route("/<id>")
|
||||
def order(id, form=None):
|
||||
def order(id: int, form: OrderForm = None) -> str:
|
||||
order = Order.query.filter(Order.id == id).first()
|
||||
if order is None:
|
||||
abort(404)
|
||||
|
@ -68,7 +63,7 @@ def order(id, form=None):
|
|||
|
||||
|
||||
@order_bp.route("/<id>/items")
|
||||
def items_showcase(id, form=None):
|
||||
def items_showcase(id: int, form: OrderForm = None) -> str:
|
||||
order = Order.query.filter(Order.id == id).first()
|
||||
if order is None:
|
||||
abort(404)
|
||||
|
@ -80,7 +75,7 @@ def items_showcase(id, form=None):
|
|||
|
||||
@order_bp.route("/<id>/edit", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def order_edit(id):
|
||||
def order_edit(id: int) -> typing.Union[str, Response]:
|
||||
order = Order.query.filter(Order.id == id).first()
|
||||
if current_user.id is not order.courrier_id and not current_user.is_admin():
|
||||
abort(401)
|
||||
|
@ -96,7 +91,9 @@ def order_edit(id):
|
|||
|
||||
|
||||
@order_bp.route("/<id>/create", methods=["POST"])
|
||||
def order_item_create(id):
|
||||
def order_item_create(id: int) -> typing.Any:
|
||||
# type is 'typing.Union[str, Response]', but this errors due to
|
||||
# https://github.com/python/mypy/issues/7187
|
||||
current_order = Order.query.filter(Order.id == id).first()
|
||||
if current_order is None:
|
||||
abort(404)
|
||||
|
@ -124,7 +121,7 @@ def order_item_create(id):
|
|||
|
||||
@order_bp.route("/<order_id>/<item_id>/paid")
|
||||
@login_required
|
||||
def item_paid(order_id, item_id):
|
||||
def item_paid(order_id: int, item_id: int) -> typing.Optional[Response]:
|
||||
item = OrderItem.query.filter(OrderItem.id == item_id).first()
|
||||
id = current_user.id
|
||||
if item.order.courrier_id == id or current_user.admin:
|
||||
|
@ -137,9 +134,9 @@ def item_paid(order_id, item_id):
|
|||
|
||||
@order_bp.route("/<order_id>/<user_name>/user_paid")
|
||||
@login_required
|
||||
def items_user_paid(order_id, user_name):
|
||||
def items_user_paid(order_id: int, user_name: str) -> typing.Optional[Response]:
|
||||
user = User.query.filter(User.username == user_name).first()
|
||||
items = []
|
||||
items: typing.List[OrderItem] = []
|
||||
if user:
|
||||
items = OrderItem.query.filter(
|
||||
(OrderItem.user_id == user.id) & (OrderItem.order_id == order_id)
|
||||
|
@ -155,13 +152,15 @@ def items_user_paid(order_id, user_name):
|
|||
for item in items:
|
||||
item.paid = True
|
||||
db.session.commit()
|
||||
flash("Paid %d items for %s" % (items.count(), item.get_name()), "success")
|
||||
flash("Paid %d items for %s" % (len(items), item.get_name()), "success")
|
||||
return redirect(url_for("order_bp.order", id=order_id))
|
||||
abort(404)
|
||||
|
||||
|
||||
@order_bp.route("/<order_id>/<item_id>/delete")
|
||||
def delete_item(order_id, item_id):
|
||||
def delete_item(order_id: int, item_id: int) -> typing.Any:
|
||||
# type is 'typing.Optional[Response]', but this errors due to
|
||||
# https://github.com/python/mypy/issues/7187
|
||||
item = OrderItem.query.filter(OrderItem.id == item_id).first()
|
||||
id = None
|
||||
if not current_user.is_anonymous():
|
||||
|
@ -178,7 +177,7 @@ def delete_item(order_id, item_id):
|
|||
|
||||
@order_bp.route("/<id>/volunteer")
|
||||
@login_required
|
||||
def volunteer(id):
|
||||
def volunteer(id: int) -> Response:
|
||||
order = Order.query.filter(Order.id == id).first()
|
||||
if order is None:
|
||||
abort(404)
|
||||
|
@ -193,7 +192,7 @@ def volunteer(id):
|
|||
|
||||
@order_bp.route("/<id>/close")
|
||||
@login_required
|
||||
def close_order(id):
|
||||
def close_order(id: int) -> typing.Optional[Response]:
|
||||
order = Order.query.filter(Order.id == id).first()
|
||||
if order is None:
|
||||
abort(404)
|
||||
|
@ -208,9 +207,13 @@ def close_order(id):
|
|||
order.courrier_id = courrier.id
|
||||
db.session.commit()
|
||||
return redirect(url_for("order_bp.order", id=id))
|
||||
# The line below is to make sure mypy doesn't say
|
||||
# "Missing return statement"
|
||||
# https://github.com/python/mypy/issues/4223
|
||||
return None
|
||||
|
||||
|
||||
def select_user(items):
|
||||
def select_user(items) -> typing.Optional[User]:
|
||||
user = None
|
||||
# remove non users
|
||||
items = [i for i in items if i.user_id]
|
||||
|
@ -228,8 +231,8 @@ def select_user(items):
|
|||
return user
|
||||
|
||||
|
||||
def get_orders(expression=None):
|
||||
orders = []
|
||||
def get_orders(expression=None) -> typing.List[Order]:
|
||||
orders: typing.List[OrderForm] = []
|
||||
if expression is None:
|
||||
expression = (datetime.now() > Order.starttime) & (
|
||||
Order.stoptime > datetime.now()
|
||||
|
|
|
@ -8,7 +8,7 @@ stats_blueprint = Blueprint("stats_blueprint", __name__)
|
|||
|
||||
|
||||
@stats_blueprint.route("/")
|
||||
def stats():
|
||||
def stats() -> str:
|
||||
data = {
|
||||
"amount": {
|
||||
"orders": FatOrder.amount(),
|
||||
|
|
16
app/zeus.py
16
app/zeus.py
|
@ -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_oauthlib.client import OAuthException, OAuth
|
||||
from flask_oauthlib.client import OAuth, OAuthException
|
||||
from werkzeug.wrappers import Response
|
||||
|
||||
from models import User, db
|
||||
|
||||
|
@ -14,7 +18,9 @@ def zeus_login():
|
|||
|
||||
|
||||
@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()
|
||||
if resp is None:
|
||||
return "Access denied: reason=%s error=%s" % (
|
||||
|
@ -60,12 +66,12 @@ def init_oauth(app):
|
|||
return zeus
|
||||
|
||||
|
||||
def login_and_redirect_user(user):
|
||||
def login_and_redirect_user(user) -> Response:
|
||||
login_user(user)
|
||||
return redirect(url_for("general_bp.home"))
|
||||
|
||||
|
||||
def create_user(username):
|
||||
def create_user(username) -> User:
|
||||
user = User()
|
||||
user.configure(username, False, 1)
|
||||
db.session.add(user)
|
||||
|
|
4
tests.md
Normal file
4
tests.md
Normal file
|
@ -0,0 +1,4 @@
|
|||
# Tests
|
||||
For this application we run a number of tests, at the moment the tests are:
|
||||
- [mypy](http://mypy-lang.org/), type-checking, which you can run using `mypy` with the argument `--ignore-missing-imports` if it says modules cannot be found
|
||||
|
Loading…
Reference in a new issue