application.py

import glob
import json
import os
from typing import NoReturn, Union

from flask import Flask, g, render_template, request, send_from_directory, session
from flask_caching import Cache
from flask_cors import CORS
from flask_login import (
    LoginManager,
    current_user,
    login_required,
    login_user,
    logout_user,
)
from flask_swagger_ui import get_swaggerui_blueprint
from werkzeug.wrappers.response import Response

import db.mongodb_controller as db_controller
import src.request_processor as request_processor
import src.utils as utils
from api.blueprint import api_blueprint
from src.config import *
from src.generate_open_api_spec import generate_api_docs
from src.models import User

# create the application object
app = Flask(__name__)
cors = CORS(app, resources={f"/api/*": {"origins": "*"}})

swaggerui_blueprint = get_swaggerui_blueprint(
    SWAGGER_URL, API_URL, config={"app_name": "RocoLib API"}
)

app.register_blueprint(swaggerui_blueprint)
app.register_blueprint(api_blueprint)

# app.config.from_pyfile('config.py')
# Might have to change how this is computed
app.secret_key = b"\xf7\x81Q\x89}\x02\xff\x98<et^"
cache = Cache(app, config={"CACHE_TYPE": "simple"})
login_manager = LoginManager(app)
login_manager.login_view = "login"


# def make_cache_key_create() -> str:
#     return (request.path + get_gym()).encode('utf-8')


@app.before_request
def open_database_connection() -> None:
    """Open DDBB connection before request so that
    it can be accessed during the request processing"""
    g.db = utils.get_db_connection()


@app.teardown_appcontext
def close_db_connection(exception) -> None:
    """Closes the database again at the end of the request."""
    if "database" in g:
        g.database.client.close()


# user loading callback
@login_manager.user_loader
def load_user(user_id: str) -> Union[User, None]:
    return User.get_by_id(user_id, utils.get_db_connection())


# Load favicon
@app.route("/favicon.ico")
def favicon() -> Response:
    """Load page favicon

    :return: the favicon file
    :rtype: Response
    """
    return send_from_directory(
        os.path.join(app.root_path, "static/images/favicon"),
        "favicon.ico",
        mimetype="image/vnd.microsoft.icon",
    )


# load languages files
LANG = {}
language_list = glob.glob("language/*.json")
for lang in language_list:
    lang_code = lang.split(os.path.sep)[1].split(".")[0]
    with open(lang, "r", encoding="utf8") as file:
        LANG[lang_code] = json.loads(file.read())


@app.context_processor
def inject_language() -> dict:
    """Inject the strings of the selected laguage based on browser config"""
    lang = utils.choose_language(request, LANG)
    return {**LANG[DEFAULT_LANG], **LANG[lang]}


def get_gym() -> str:
    """Get the current session's selected gym."""
    return utils.get_current_gym(session, g.db)


@app.route("/", methods=["GET", "POST"])
def home() -> str:
    return request_processor.handle_home_request(request, session, g.db)


@app.route("/create")
# @cache.cached(timeout=60 * 60, key_prefix=make_cache_key_create)
def create() -> str:
    return request_processor.handle_create_request(request, session, g.db, current_user)


@app.route("/create_boulder")
def create_boulder() -> str:
    return render_template(
        "create_boulder.html",
        latest_set=utils.get_show_only_latest_wall_sets(current_user),
    )


@app.route("/create_route")
def create_route() -> str:
    return render_template("create_route.html")


@app.route("/explore")
def explore() -> str:
    return render_template(
        "explore.html",
        walls=db_controller.get_gym_walls(
            get_gym(), g.db, utils.get_show_only_latest_wall_sets(current_user)
        ),
    )


@app.route("/explore_boulders", methods=["GET", "POST"])
def explore_boulders() -> str:
    return request_processor.handle_explore_boulders(
        request, session, g.db, current_user
    )


@app.route("/explore_circuits", methods=["GET", "POST"])
def explore_circuits() -> str:
    return request_processor.handle_explore_circuits(
        request, session, g.db, current_user
    )


@app.route("/change_gym", methods=["POST"])
def change_gym_problem_list() -> str:
    return request_processor.handle_change_gym_problem_list_request(
        request, session, g.db, current_user
    )


@app.route("/rate_boulder", methods=["POST"])
def rate_boulder() -> Union[Response, NoReturn]:
    return request_processor.process_rate_boulder_request(request, session, g.db)


@app.route("/load_circuit", methods=["POST", "GET"])
# @cache.cached(timeout=60*60, key_prefix=make_cache_key_boulder)
def load_circuit() -> Union[str, NoReturn]:
    return request_processor.process_load_circuit_request(
        request, session, g.db, current_user, app.static_folder
    )


@app.route("/load_boulder", methods=["POST", "GET"])
# @cache.cached(timeout=60*60, key_prefix=make_cache_key_boulder)
def load_boulder() -> Union[str, NoReturn]:
    return request_processor.process_load_boulder_request(
        request, session, g.db, current_user, app.static_folder
    )


@app.route("/load_next")
def load_next_problem():
    return request_processor.process_load_next_problem_request(
        request, session, g.db, current_user, app.static_folder
    )


@app.route("/load_previous")
def load_previous_problem():
    return request_processor.process_load_previous_problem_request(
        request, session, g.db, current_user, app.static_folder
    )


@app.route("/explore_routes")
def explore_routes() -> str:
    return render_template("explore_routes.html")


@app.route("/random_problem")
def random_problem() -> str:
    return request_processor.process_random_problem_request(
        request, session, g.db, current_user, app.static_folder
    )


@app.route("/about_us")
def render_about_us() -> str:
    return render_template("about_us.html")


@app.route("/walls/<string:wall_section>")
def wall_section(wall_section) -> str:
    return request_processor.process_wall_section_request(
        request, session, g.db, current_user, app.static_folder, wall_section
    )


@app.route("/save", methods=["POST"])
def save() -> Response:
    return request_processor.process_save_request(request, session, g.db)


@app.route("/circuit_save", methods=["POST"])
def circuit_save() -> Response:
    return request_processor.process_save_request(
        request, session, g.db, is_circuit=True
    )


@app.route("/save_boulder", methods=["POST"])
def save_boulder() -> Union[str, NoReturn]:
    return request_processor.process_save_boulder_request(request, current_user)


@app.route("/save_circuit", methods=["POST"])
def save_circuit() -> Union[str, NoReturn]:
    return request_processor.process_save_circuit_request(request, current_user)


@app.route("/add_gym", methods=["GET", "POST"])
@login_required
def add_gym() -> str:
    return render_template("add_new_gym.html")


# Login handlers
@app.route("/login", methods=["GET", "POST"])
def login() -> Union[str, Response]:
    return request_processor.process_login_request(
        request, session, g.db, current_user, login_user
    )


@app.route("/signup/", methods=["GET", "POST"])
def show_signup_form() -> Union[str, Response]:
    return request_processor.process_signup_request(
        request, session, g.db, current_user, login_user
    )


@app.route("/logout")
def logout() -> Response:
    return request_processor.process_logout_request(logout_user)


# User related endpoints


@app.route("/tick_list", methods=["GET", "POST"])
@login_required
def tick_list() -> Union[str, Response]:
    return request_processor.process_ticklist_request(
        request, session, g.db, current_user
    )


@app.route("/profile", methods=["GET", "POST"])
@login_required
def profile() -> Union[str, Response]:
    return request_processor.process_profile_request(
        request, g.db, session, current_user
    )


@app.route("/delete_ticklist_problem", methods=["POST"])
@login_required
def delete_ticklist_problem() -> Union[Response, NoReturn]:
    return request_processor.process_delete_ticklist_problem_request(
        request, g.db, current_user
    )


@app.route("/get_nearest_gym", methods=["POST"])
def get_nearest_gym() -> Response:
    return request_processor.process_get_nearest_gym_request(request, session, g.db)


@app.route("/contact", methods=["GET"])
def show_contact() -> Union[Response, NoReturn]:
    return render_template("contact.html")


@app.errorhandler(404)
def page_not_found(error) -> tuple[str, int]:
    # pylint: disable=no-member
    app.logger.error("Page not found: %s", (request.path))
    return render_template("errors/404.html"), 404


@app.errorhandler(400)
def bad_request(error) -> tuple[str, int]:
    # pylint: disable=no-member
    app.logger.error("Bad request: %s", (request.path))
    return render_template("errors/400.html"), 400


# start the server
if __name__ == "__main__":
    if GENERATE_API_DOCS:
        generate_api_docs(app)
    if RUN_SERVER:
        if DOCKER_ENV == "True":
            utils.set_creds_file(CREDS_DEV)
            app.run(debug=DEBUG, host="0.0.0.0", port=80)
        else:
            utils.set_creds_file(CREDS)
            app.run(debug=DEBUG, port=PORT)