Feature/weekly tasks

app/database/models.py

@@ -37,6 +37,12 @@ class User(Base):
 
     events = relationship("UserEvent", back_populates="participants")
 
+    weekly_tasks = relationship(
+        "WeeklyTask", cascade="all, delete", back_populates="owner")
+
+    tasks = relationship(
+        "Task", cascade="all, delete", back_populates="owner")
+
     def __repr__(self):
         return f'<User {self.id}>'
 
@@ -110,6 +116,34 @@ def __repr__(self):
         )
 
 
+class Task(Base):
+    __tablename__ = "tasks"
+
+    id = Column(Integer, primary_key=True, index=True)
+    title = Column(String)
+    content = Column(String)
+    is_done = Column(Boolean, nullable=False)
+    is_important = Column(Boolean, nullable=False)
+    date_time = Column(DateTime, nullable=False)
+    owner_id = Column(Integer, ForeignKey("users.id"))
+
+    owner = relationship("User", back_populates="tasks")
+
+
+class WeeklyTask(Base):
+    __tablename__ = "weekly_task"
+
+    id = Column(Integer, primary_key=True, index=True)
+    title = Column(String)
+    days = Column(String, nullable=False)
+    content = Column(String)
+    is_important = Column(Boolean, nullable=False)
+    the_time = Column(String, nullable=False)
+    owner_id = Column(Integer, ForeignKey("users.id"))
+
+    owner = relationship("User", back_populates="weekly_tasks")
+
+
 class Quote(Base):
     __tablename__ = "quotes"
 

app/internal/weekly_tasks.py

@@ -0,0 +1,138 @@
+from datetime import date, datetime, time
+from typing import Tuple
+
+from app.database.models import User, Task, WeeklyTask
+from sqlalchemy.orm.session import Session
+
+
+def check_inputs(days: str, the_time: time, title: str) -> bool:
+    """Checks inputs, used by the make_or_change_weekly_task function"""
+    if not days or days == "" or not the_time or not title or title == "":
+        return False
+    return True
+
+
+def make_or_change_weekly_task(
+    user: User, session: Session,
+    mode: str, weekly_task_id: int,
+    title: str, days: str,
+    content: str, is_important: bool, the_time: time
+) -> Tuple[bool, WeeklyTask]:
+    """This function is being used to add a Weekly Task to the user
+        or to edit an existing Weekly Task the user have.
+
+    Args:
+        user (User): The user who wants to add or edit the Weekly Task.
+        session (Session): The session to redirect to the database.
+        mode (str): Determines whether in Add or Edit mode.
+        weekly_task_id (int): In edit mode, represents the
+            weekly task being edited.
+        title (str): Title of the Weekly Task.
+        days (str): Return days of the Weekly Task.
+        content (str): Content of the Weekly Task.
+        is_important (bool): If the task is important.
+        the_time (time): Return time of the Weekly Task.
+
+    Returns:
+        Tuple: Boolean variable which shows if the change has been
+            made to the db, and the model WeeklyTask which the
+            function made so far.
+    """
+    weekly_task = WeeklyTask(
+        title=title,
+        content=content,
+        is_important=is_important,
+        owner_id=user.id
+    )
+
+    if weekly_task_id:
+        weekly_task.id = weekly_task_id
+
+    inputs_ok = check_inputs(days, the_time, title)
+    if not inputs_ok:
+        return False, weekly_task
+
+    weekly_task.days = days
+    weekly_task.the_time = the_time.strftime("%H:%M")
+
+    if mode == "add":
+        user_titles = (
+            user_weekly_task.title
+            for user_weekly_task in user.weekly_tasks
+            )
+        if title in user_titles:
+            return False, weekly_task
+        session.add(weekly_task)
+        session.commit()
+        return True, weekly_task
+
+    else:  # if mode == "edit"
+        w_task_query = session.query(WeeklyTask)
+        old_weekly_task = w_task_query.filter_by(id=weekly_task.id).first()
+
+        user_titles = (
+            user_weekly_task.title for user_weekly_task in user.weekly_tasks
+            if user_weekly_task.title != old_weekly_task.title
+        )
+
+        if title in user_titles:
+            return False, weekly_task
+
+        if old_weekly_task.owner_id != weekly_task.owner_id:
+            return False, weekly_task
+
+        old_weekly_task.title = weekly_task.title
+        old_weekly_task.days = weekly_task.days
+        old_weekly_task.content = weekly_task.content
+        old_weekly_task.is_important = weekly_task.is_important
+        old_weekly_task.the_time = weekly_task.the_time
+        session.commit()
+        return True, weekly_task
+
+
+def make_task(task: Task, user: User, session: Session) -> bool:
+    """Make a task, used by the generate_tasks function"""
+    user_tasks_query = session.query(Task).filter_by(owner_id=user.id)
+    task_by_time = user_tasks_query.filter_by(date_time=task.date_time)
+    task_by_title_and_time = task_by_time.filter_by(title=task.title)
+    task_exist = task_by_title_and_time.first()
+    if not task_exist:
+        session.add(task)
+        session.commit()
+        return True
+    return False
+
+
+def generate_tasks(session: Session, user: User):
+    """Generates tasks for the week
+    based on all the weekly tasks the user have"""
+    current_date = date.today()
+    current_week_num = current_date.strftime("%W")
+    current_year = current_date.strftime("%Y")
+
+    for weekly_task in user.weekly_tasks:
+        the_time = weekly_task.the_time
+        days = weekly_task.days.split(", ")
+        for day in days:
+            date_string = f"{day} {the_time} {current_week_num} {current_year}"
+            date_time = datetime.strptime(date_string, "%a %H:%M %W %Y")
+            task = Task(
+                title=weekly_task.title,
+                content=weekly_task.content,
+                is_done=False,
+                is_important=weekly_task.is_important,
+                date_time=date_time,
+                owner_id=user.id
+            )
+            make_task(task, user, session)
+
+
+def remove_weekly_task(weekly_task_id: int, session: Session) -> bool:
+    """Removes a weekly task from the db based on the weekly task id"""
+    weekly_task_query = session.query(WeeklyTask)
+    weekly_task = weekly_task_query.filter_by(id=weekly_task_id).first()
+    if weekly_task:
+        session.query(WeeklyTask).filter_by(id=weekly_task_id).delete()
+        session.commit()
+        return True
+    return False

app/main.py

@@ -7,10 +7,11 @@
 from app.database.database import engine, get_db
 from app.dependencies import (
     MEDIA_PATH, STATIC_PATH, templates)
+
 from app.internal.quotes import load_quotes, daily_quotes
 from app.routers import (
     agenda, dayview, email, event, invitation, profile, search, telegram,
-    whatsapp
+    weekly_tasks, whatsapp
     )
 from app.telegram.bot import telegram_bot
 from app.internal.logger_customizer import LoggerCustomizer
@@ -51,6 +52,7 @@ def create_tables(engine, psql_environment):
 app.include_router(dayview.router)
 app.include_router(email.router)
 app.include_router(invitation.router)
+app.include_router(weekly_tasks.router)
 app.include_router(whatsapp.router)
 app.include_router(search.router)
 

app/routers/weekly_tasks.py

@@ -0,0 +1,153 @@
+import datetime
+
+from fastapi import APIRouter, Depends, Request, Form
+from starlette.responses import RedirectResponse
+from starlette.status import HTTP_302_FOUND
+
+from app.database.database import get_db
+from app.database.models import User, WeeklyTask
+from app.dependencies import templates
+from app.internal.weekly_tasks import (
+    make_or_change_weekly_task, remove_weekly_task)
+
+
+router = APIRouter(
+    prefix="/weekly-tasks",
+    tags=["weekly-tasks"],
+    responses={404: {"description": "Not found"}},
+)
+
+
+def get_placeholder_user():
+    user = User(
+        username='demo_user',
+        email='[email protected]',
+        password='s3jsd183b13',
+        full_name='The Demo'
+    )
+    return user
+
+
+def get_user(demo_user, session):
+    user = session.query(User).filter_by(username=demo_user.username).first()
+    if not user:
+        session.add(demo_user)
+        session.commit()
+        user = session.query(User).filter_by(id=1).first()
+    return user
+
+
+@router.get("/")
+async def weekly_tasks_manager(
+        request: Request,
+        session=Depends(get_db),
+        demo_user=Depends(get_placeholder_user)):
+
+    user = get_user(demo_user, session)
+
+    # TODO: Move the below function to a compatible location
+    # Need to run regularly whenever there are no tasks on the week
+    # Or will run on the background after the user left the
+    # weekly-tasks manager page
+    # function:
+    # generate_tasks(session, user)  # imported from app.internal.weekly_tasks
+    # session.close()
+
+    return templates.TemplateResponse("weekly_tasks_manager.html", {
+        "request": request,
+        "weekly_tasks": user.weekly_tasks
+    })
+
+
+@router.get("/add")
+async def weekly_task_add(
+        request: Request):
+
+    return templates.TemplateResponse("add_edit_weekly_task.html", {
+        "request": request,
+        "weekly_task": None,
+        "mode": "add"
+    })
+
+
+@router.post("/remove")
+async def weekly_task_remove(
+        session=Depends(get_db),
+        remove_id: int = Form(...)):
+
+    remove_weekly_task(remove_id, session)
+    session.close()
+    url = router.url_path_for("weekly_tasks_manager")
+    return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
+
+
+@router.post("/edit")
+async def weekly_task_edit(
+        request: Request,
+        session=Depends(get_db),
+        edit_id: int = Form(...)):
+
+    weekly_task = session.query(WeeklyTask).filter_by(id=edit_id).first()
+    return templates.TemplateResponse("add_edit_weekly_task.html", {
+        "request": request,
+        "weekly_task": weekly_task,
+        "mode": "edit"
+    })
+
+
+@router.post("/make-change")
+async def weekly_task_make_change(
+        request: Request,
+        session=Depends(get_db),
+        demo_user=Depends(get_placeholder_user),
+        title: str = Form(None),
+        sun: bool = Form(False),
+        mon: bool = Form(False),
+        tue: bool = Form(False),
+        wed: bool = Form(False),
+        thu: bool = Form(False),
+        fri: bool = Form(False),
+        sat: bool = Form(False),
+        content: str = Form(None),
+        is_important: bool = Form(False),
+        the_time: datetime.time = Form(None),
+        weekly_task_id: int = Form(...),
+        mode: str = Form(...)):
+
+    user = get_user(demo_user, session)
+    days_dict = {
+        "Sun": sun,
+        "Mon": mon,
+        "Tue": tue,
+        "Wed": wed,
+        "Thu": thu,
+        "Fri": fri,
+        "Sat": sat
+    }
+    days_list = [day for day, is_true in days_dict.items() if is_true]
+    days = ", ".join(days_list)
+
+    made_change, weekly_task = make_or_change_weekly_task(
+        user, session,
+        mode, weekly_task_id,
+        title, days,
+        content, is_important,
+        the_time,
+    )
+
+    massage = None
+    if mode == "add":
+        massage = "could not add The Weekly Task"
+    else:
+        massage = "These changes could not be made to the Weekly Task"
+
+    if not made_change:
+        return templates.TemplateResponse("add_edit_weekly_task.html", {
+            "request": request,
+            "massage": massage,
+            "weekly_task": weekly_task,
+            "mode": mode
+        })
+    session.close()
+    url = router.url_path_for("weekly_tasks_manager")
+    return RedirectResponse(url=url, status_code=HTTP_302_FOUND)