123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
from flask import Flask, make_response
|
|
from flask import g
|
|
from flask_mail import Mail
|
|
from flask_security import Security, SQLAlchemyUserDatastore, user_registered
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
import os
|
|
from config import config
|
|
from flask_security.forms import LoginForm
|
|
from flask_limiter import Limiter
|
|
from flask_limiter.util import get_remote_address
|
|
|
|
from .forms import *
|
|
|
|
|
|
app = Flask(__name__)
|
|
app.config.from_object(config[os.getenv('FLASK_CONFIG') or 'default'])
|
|
|
|
# applies to all routes, so choose limits wisely!
|
|
limiter = Limiter(
|
|
app,
|
|
key_func=get_remote_address,
|
|
# default_limits=["500 per second"]
|
|
)
|
|
|
|
|
|
@app.errorhandler(429)
|
|
def ratelimit_handler(e):
|
|
return make_response(
|
|
jsonify(error="ratelimit exceeded %s" % e.description)
|
|
, 429
|
|
)
|
|
db = SQLAlchemy(app)
|
|
mail = Mail(app)
|
|
|
|
from .entities import *
|
|
|
|
user_datastore = SQLAlchemyUserDatastore(db, User, Role)
|
|
security = Security(app, user_datastore)
|
|
|
|
# required to activate the filters
|
|
from .filters import *
|
|
from .tools import *
|
|
from .routes import *
|
|
|
|
|
|
@user_registered.connect_via(app)
|
|
def user_registered_sighandler(application, user, confirm_token):
|
|
"""
|
|
Called after a user was created
|
|
"""
|
|
role = user_datastore.find_role('user')
|
|
user_datastore.add_role_to_user(user, role)
|
|
if user.email == application.config['ADMIN_MAIL']:
|
|
# if the user selected the preconfigured email for the admin account
|
|
role = user_datastore.find_role('admin')
|
|
user_datastore.add_role_to_user(user, role)
|
|
new_vehicle = Vehicle('default vehicle')
|
|
db.session.add(new_vehicle)
|
|
user.vehicles.append(new_vehicle)
|
|
db.session.commit()
|
|
tools.db_log_add(user)
|
|
tools.db_log_add(new_vehicle)
|
|
|
|
|
|
def assure_consumable(name, ext_id, unit):
|
|
if not Consumable.query.filter(Consumable.ext_id == ext_id).first():
|
|
c = Consumable(name, ext_id, unit)
|
|
db.session.add(c)
|
|
|
|
|
|
@app.before_first_request
|
|
def before_first_request():
|
|
db.create_all()
|
|
|
|
# make sure all consumables from tankerkoenig exist: diesel, e5, e10
|
|
assure_consumable('Diesel', 'diesel', 'L')
|
|
assure_consumable('Super','e5', 'L')
|
|
assure_consumable('Super E10','e10', 'L')
|
|
|
|
user_datastore.find_or_create_role(name='admin', description='Role for administrators')
|
|
user_datastore.find_or_create_role(name='user', description='Role for all users.')
|
|
db.session.commit()
|
|
|
|
|
|
@app.before_request
|
|
def before_request():
|
|
g.data = {}
|
|
|
|
|
|
@app.route('/')
|
|
def index():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('get_pit_stops'))
|
|
else:
|
|
user_count = len(User.query.all())
|
|
consumables = Consumable.query.all()
|
|
per_consumable = {}
|
|
for consumable in consumables:
|
|
per_consumable[consumable.id] = {
|
|
'name': consumable.name,
|
|
'unit': consumable.unit,
|
|
'amount': 0
|
|
}
|
|
vehicles = Vehicle.query.all()
|
|
kilometers = 0
|
|
for vehicle in vehicles:
|
|
stats = tools.VehicleStats(vehicle)
|
|
for consumable in stats.consumables:
|
|
per_consumable[consumable.id]['amount'] += consumable.overall_amount
|
|
kilometers += stats.overall_distance
|
|
vehicle_count = len(vehicles)
|
|
pitstop_count = len(Pitstop.query.all())
|
|
data = {
|
|
'users': user_count,
|
|
'vehicles': vehicle_count,
|
|
'pitstops': pitstop_count,
|
|
'kilometers': kilometers,
|
|
'consumables': per_consumable
|
|
}
|
|
|
|
return render_template('index.html', login_user_form=LoginForm(), data=data)
|
|
|