|
|
@@ -0,0 +1,170 @@
|
|
|
+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")
|
|
|
+
|
|
|
+
|
|
|
[email protected]("/")
|
|
|
+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)
|
|
|
+
|
|
|
+
|
|
|
[email protected]("/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")
|
|
|
+
|
|
|
+
|
|
|
[email protected]("/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"))
|
|
|
+
|
|
|
+
|
|
|
[email protected]("/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)
|