import json
from flask import (
Flask,
redirect,
render_template,
request,
)
from database import use_database
from cookie import create_access_token, get_token_payload
from utils import redirect_if_logged_in, admin_only
# number of days until token expires
TOKEN_DURATION = 1
# initialize a Flask application
app = Flask(__name__)
# route for the homepage
@app.route("/")
@redirect_if_logged_in("/dashboard") # redirect to dashboard if user is logged in
def home():
response = redirect("/login") # redirect to login page
response.delete_cookie("token") # delete token cookie
return response
# route for the login page
@app.route("/login")
@redirect_if_logged_in("/dashboard") # redirect to dashboard if user is logged in
def login_page():
return render_template("login.html") # render the login page template
# route for the dashboard page
@app.route("/dashboard")
@use_database("database.db") # use the database context manager
def dashboard_page(database):
print(request.cookies.get("token")) # print the token cookie to the console
username = get_token_payload(request.cookies.get("token"))["username"] # get the username from the token payload
user = database.get_user(username) # get the user from the database
return render_template("home.html", user=user) # render the dashboard template with the user object
# route for the users page
@app.route("/dashboard/users")
@use_database("database.db") # use the database context manager
def users_page(database):
username = get_token_payload(request.cookies.get("token"))["username"] # get the username from the token payload
user = database.get_user(username) # get the user from the database
return render_template("users.html", user=user) # render the users template with the user object
# route for the login API
@app.route("/api/login", methods=["POST"])
@use_database("database.db") # use the database context manager
def login_handler(database):
try:
data = request.get_json() # get the JSON data from the request
username, password = data["username"], data["password"] # extract the username and password from the JSON data
except KeyError:
return json.dumps({
"status": "error",
"message": "Missing fields",
})
if not all([username, password]): # check if both fields are present
return json.dumps({
"status": "error",
"message": "Missing fields",
})
if database.verify_password(username, password): # check if the username and password match
token = create_access_token({"username": username}, TOKEN_DURATION) # create a token with the username as the payload
response = redirect("/dashboard") # redirect to the dashboard page
response.set_cookie("token", token, path="/") # set the token cookie
response.set_data(json.dumps({
"status": "success",
"message": "Logged in",
}))
return response # return the response with the token cookie and JSON message
# This route handles user registration
@app.route("/api/register", methods=["POST"])
@use_database("database.db")
def register_handler(database):
try:
data = request.get_json()
email, username, password = data["email"], data["username"], data["password"]
except KeyError:
return json.dumps({
"status": "error",
"message": "Missing fields",
})
# Check if all the required fields are present in the request body
if not all([email, username, password]):
return json.dumps({
"status": "error",
"message": "Missing fields",
})
# Check if the user already exists in the database
if database.get_user(username) is None:
# Register the user
database.register_user(email, username, password, "user")
return json.dumps({
"status": "success",
"message": "User registered",
})
# Return an error if the user already exists
return json.dumps({
"status": "error",
"message": "User already exists",
})
# This route handles user logout
@app.route("/api/logout", methods=["POST"])
def logout_handler():
# Delete the access token cookie and redirect the user to the login page
response = redirect("/login")
response.delete_cookie("token")
response.set_data(json.dumps({
"status": "success",
"message": "Logged out",
}))
return response
# This route returns a list of all users
@app.route("/api/users", methods=["GET"])
@use_database("database.db")
@admin_only()
def get_users(database, admin_user):
# Get all users from the database and map them to a dictionary format
users = database.get_all_users()
users = map(lambda user: {
"id": user[0],
"email": user[1],
"username": user[2],
"password": user[3],
"role": user[4],
}, users)
# Return the list of users as a JSON response
return json.dumps({
"status": "success",
"users": list(users),
})
# This route updates a user's information
@app.route("/api/users", methods=["UPDATE"])
@use_database("database.db")
@admin_only()
def update_users(database, admin_user):
try:
# Get the user information from the request body
data = request.get_json()
user_id, email, username, password, role = data["id"], data["email"], data["username"], data["password"], data["role"]
except KeyError:
return json.dumps({
"status": "error",
"message": "Missing fields",
})
# Update the user's information in the database
success = database.update_user(
user_id,
email,
username,
password,
role
)
# Return a success or error message as a JSON response
if success:
return json.dumps({
"status": "success",
})
return json.dumps({
"status": "error",
"message": "Unable to update user. Please check console.",
})
# This route deletes a user from the database
@app.route("/api/users", methods=["DELETE"])
@use_database("database.db")
@admin_only()
def delete_users(database, admin_user):
try:
# Get the user ID from the request body
data = request.get_json()
user_id = data["id"]
except KeyError:
return json.dumps({
"status