Toppers-Tasks/main.py

from email.policy import default
import os
import re
import bcrypt
import dotenv
from flask import (
    Flask,
    flash,
    redirect,
    render_template,
    request,
    session,
    url_for,
)
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.sql import func


dotenv.load_dotenv()
app = Flask(__name__)
app.secret_key = os.getenv("SECRET_KEY", "whataloadofnonsense")


class Base(DeclarativeBase):
    pass


app.config["SQLALCHEMY_DATABASE_URI"] = (
    "mysql://root:root@localhost:3306/topperstasks"
)
db = SQLAlchemy(model_class=Base)
db.init_app(app)


class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String, unique=True, nullable=False)
    passhash = db.Column(db.String, nullable=False)
    is_admin = db.Column(db.Boolean, default=False)

    tasks = db.relationship("Task", back_populates="user")
    task_assignments = db.relationship("Task_Assignment", back_populates="user")


class Task(db.Model):
    __tablename__ = "tasks"
    id = db.Column(db.Integer, primary_key=True)
    created_datetime = db.Column(db.String, nullable=False, default=func.now())
    tasktext = db.Column(db.String, nullable=False)
    created_by = db.Column(db.Integer, db.ForeignKey(User.id), nullable=False)
    due = db.Column(db.String)
    completed = db.Column(db.Boolean, default=False)
    deleted = db.Column(db.Boolean, default=False)

    user = db.relationship(User, back_populates="tasks")
    task_assignments = db.relationship("Task_Assignment", back_populates="task")


class Task_Assignment(db.Model):
    __tablename__ = "task_assignments"
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey(User.id), nullable=False)
    task_id = db.Column(db.Integer, db.ForeignKey(Task.id), nullable=False)

    user = db.relationship(User, back_populates="task_assignments")
    task = db.relationship(Task, back_populates="task_assignments")


@app.route("/")
def home():
    if not session.get("userid"):
        return redirect(url_for("login"))

    tasks = db.session.execute(
        db.select(Task).where(Task.deleted == False)
    ).scalars()

    return render_template("home.html", tasks=tasks)


@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        try:
            username = request.form.get("user")
            password = request.form.get("password")

            user: User = db.session.execute(
                db.select(User).where(User.username == username)
            ).scalar_one_or_none()

            if not user:
                flash("Failed to login", category="warning")
                return redirect(url_for("login"))

            if not user.passhash:
                salt = bcrypt.gensalt()
                user.passhash = bcrypt.hashpw(password.encode(), salt)
                db.session.commit()
                flash("Password set succesfully!", category="success")
                return redirect(url_for("login"))

            if bcrypt.checkpw(password.encode(), user.passhash.encode()):
                session["userid"] = user.id
                return redirect(url_for("home"))
            else:
                flash("Failed to login", category="warning")
                return redirect(url_for("login"))
        except:
            flash("Critical Error, contact Peter pls", category="error")
            return redirect(url_for("login"))

    return render_template("login.html")


@app.route("/addtask", methods=["POST"])
def addtask():
    try:
        tasktext = request.form.get("task-text")
        user_id = session.get("userid")

        if not tasktext:
            flash("Invalid task text", category="warning")
            return redirect(url_for("home"))

        if not user_id:
            return redirect(url_for("login"))

        newtask = Task(tasktext=tasktext, created_by=user_id)
        db.session.add(newtask)
        db.session.commit()
    except:
        flash("Critical Error, contact Peter pls", category="error")
        return redirect(url_for("home"))

    return redirect(url_for("home"))


@app.route("/updatetask", methods=["POST"])
def updatetask():
    if "task-id" not in request.form:
        flash("Failed to find task", category="warning")
        return redirect(url_for("home"))

    task = db.session.execute(
        db.select(Task).where(Task.id == request.form.get("task-id"))
    ).scalar_one_or_none()

    if not task:
        flash("Failed to find task", category="warning")
        return redirect(url_for("home"))

    if "due" in request.form:
        if not task.due:
            flash("Invalid due date", category="warning")
            return redirect(url_for("home"))

        datetime_str = request.form.get("due")
        datetime_str = datetime_str.replace("T", " ")
        datetime_str += ":00"
        task.due = datetime_str

        db.session.commit()
        return redirect(url_for("home"))


if __name__ == "__main__":
    app.run(host="0.0.0.0", debug=True)