Implement the Task Manager

projects/task_manager/.gitignore

@@ -0,0 +1,6 @@
+.idea/
+.venv/
+__pycache__/
+.DS_Store
+
+instance/

projects/task_manager/README.MD

@@ -0,0 +1,23 @@
+# Personal Task Manager
+
+В данном проекте реализован проект менеджера задач согласно [описанию](./task_description.md). В качестве базы данных
+используется встроенный в стандартную библиотеку модуль `sqlite3`, а для пользовательского интерфейса реализовано
+веб-приложение на [Flask](https://github.com/pallets/flask).
+
+## Запуск
+
+Для первого запуска приложения рекомендуется создать виртуальное окружение и установить зависимости:
+
+```shell
+python3 -m venv .venv
+source .venv/bin/activate
+python3 -m pip install -r requirements.txt
+```
+
+Далее для запуска веб-приложение необходимо вызвать:
+
+```shell
+python3 -m flask run
+```
+
+Актуальный адрес для подключения к приложению будет выведен в стандартный вывод. По умолчанию это http://127.0.0.1:5000. 

projects/task_manager/app.py

@@ -0,0 +1,25 @@
+import os
+import sys
+import sqlite3
+from flask import Flask
+from routes import tasks_bp
+
+app = Flask(__name__)
+app.register_blueprint(tasks_bp)
+app.config.from_mapping(
+    DATABASE=os.path.join(app.instance_path, 'tasks.sqlite')
+)
+
+
+@app.errorhandler(sqlite3.Error)
+def handle_error(err):
+    return f'Database operation failed: {err}', 500
+
+
+try:
+    os.makedirs(app.instance_path, exist_ok=True)
+except OSError as e:
+    sys.exit(f'Failed to create the instance directory at {e.filename}')
+
+if __name__ == '__main__':
+    app.run()

projects/task_manager/managers/csv_manager.py

@@ -0,0 +1,36 @@
+import csv
+from .task_manager import TaskManager, TaskPriority
+
+
+class CSVManager:
+    def __init__(self):
+        self.task_manager = TaskManager.get_instance()
+
+    def import_tasks(self, file_path):
+        with open(file_path, 'r') as file:
+            reader = csv.DictReader(file)
+            for row in reader:
+                title = row['Title']
+                description = row['Description']
+                deadline = row['Deadline']
+                priority = TaskPriority[row['Priority']]
+                project = row['Project']
+
+                self.task_manager.add_task(title, description, deadline, priority, project)
+
+    def export_tasks(self, file_path):
+        tasks = self.task_manager.get_tasks(sort_by=None, order=None, project=None)
+
+        with open(file_path, 'w') as file:
+            fieldnames = ['Title', 'Description', 'Deadline', 'Priority', 'Project']
+            writer = csv.DictWriter(file, fieldnames=fieldnames)
+            writer.writeheader()
+
+            for task in tasks:
+                writer.writerow({
+                    'Title': task[1],
+                    'Description': task[2],
+                    'Deadline': task[3],
+                    'Priority': task[4],
+                    'Project': task[5]
+                })

projects/task_manager/managers/task_manager.py

@@ -0,0 +1,117 @@
+import sqlite3
+from enum import Enum, auto
+from flask import current_app, g
+
+
+class TaskPriority(Enum):
+    LOW = auto()
+    MEDIUM = auto()
+    HIGH = auto()
+
+
+class TaskManager:
+    def __init__(self):
+        db_name = current_app.config['DATABASE']
+        self.conn = sqlite3.connect(db_name)
+        self.create_table()
+
+    def __del__(self):
+        self.conn.close()
+
+    def create_table(self):
+        cursor = self.conn.cursor()
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS tasks (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                title TEXT NOT NULL,
+                description TEXT,
+                deadline DATE,
+                priority TEXT CHECK(priority IN ('LOW', 'MEDIUM', 'HIGH')),
+                project TEXT
+            )
+        ''')
+        self.conn.commit()
+
+    def add_task(self, title, description, deadline, priority, project):
+        TaskManager.__validate_priority(priority)
+
+        cursor = self.conn.cursor()
+        cursor.execute('''
+            INSERT INTO tasks (title, description, deadline, priority, project)
+            VALUES (?, ?, ?, ?, ?)
+        ''', (title, description, deadline, priority.name, project))
+        self.conn.commit()
+
+    def get_tasks(self, sort_by, order, project):
+        if sort_by not in ('id', 'title', 'description', 'deadline', 'priority'):
+            sort_by = 'id'
+        if order not in ('ASC', 'DESC'):
+            order = 'ASC'
+
+        query = ['SELECT * FROM tasks']
+        values = []
+
+        if project:
+            query.append('WHERE project = ?')
+            values.append(project)
+
+        query.append(f'ORDER BY {sort_by} {order}')
+
+        cursor = self.conn.cursor()
+        cursor.execute(' '.join(query), values)
+        return cursor.fetchall()
+
+    def get_task_by_id(self, task_id):
+        cursor = self.conn.cursor()
+        cursor.execute('SELECT * FROM tasks WHERE id = ?', (task_id,))
+        return cursor.fetchone()
+
+    def update_task(self, task_id, title=None, description=None, deadline=None, priority=None, project=None):
+        update_query = 'UPDATE tasks SET '
+        updates = []
+        values = []
+
+        if title:
+            updates.append('title = ?')
+            values.append(title)
+        if description:
+            updates.append('description = ?')
+            values.append(description)
+        if deadline:
+            updates.append('deadline = ?')
+            values.append(deadline)
+        if priority:
+            TaskManager.__validate_priority(priority)
+            updates.append('priority = ?')
+            values.append(priority.name)
+        if project:
+            updates.append('project = ?')
+            values.append(project)
+
+        update_query += ', '.join(updates) + ' WHERE id = ?'
+        values.append(task_id)
+
+        cursor = self.conn.cursor()
+        cursor.execute(update_query, values)
+        self.conn.commit()
+
+    def delete_task(self, task_id):
+        cursor = self.conn.cursor()
+        cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,))
+        self.conn.commit()
+
+    def get_projects(self):
+        cursor = self.conn.cursor()
+        cursor.execute('SELECT DISTINCT project FROM tasks ORDER BY project ASC')
+        return [project[0] for project in cursor.fetchall()]
+
+    @staticmethod
+    def get_instance():
+        if 'task_manager' not in g:
+            g.task_manager = TaskManager()
+        return g.task_manager
+
+    @staticmethod
+    def __validate_priority(priority):
+        if not isinstance(priority, TaskPriority):
+            raise ValueError('Priority must be of type "TaskPriority"')

projects/task_manager/requirements.txt

@@ -0,0 +1 @@
+Flask==3.0.0

projects/task_manager/routes.py

@@ -0,0 +1,111 @@
+import os
+import sys
+from tempfile import gettempdir
+
+from managers.csv_manager import CSVManager
+from managers.task_manager import TaskManager, TaskPriority
+from flask import Blueprint, render_template, request, redirect, url_for, send_file, after_this_request
+
+tasks_bp = Blueprint('tasks', __name__, template_folder='templates')
+
+
+@tasks_bp.route('/')
+@tasks_bp.route('/tasks')
+def show_tasks():
+    sort_by = request.args.get('sort_by', 'id')
+    order = request.args.get('order', 'ASC')
+    project = request.args.get('project')
+
+    task_manager = TaskManager.get_instance()
+    tasks = task_manager.get_tasks(sort_by, order, project)
+    projects = task_manager.get_projects()
+
+    return render_template('task_list.html', tasks=tasks, projects=projects)
+
+
+@tasks_bp.route('/add_task', methods=['GET', 'POST'])
+def add_task():
+    if request.method == 'POST':
+        title = request.form['title']
+        description = request.form['description']
+        deadline = request.form['deadline']
+        priority = TaskPriority[request.form['priority']]
+        project = request.form['project']
+
+        task_manager = TaskManager.get_instance()
+        task_manager.add_task(title, description, deadline, priority, project)
+        return redirect(url_for('.show_tasks'))
+    return render_template('task_form.html')
+
+
+@tasks_bp.route('/update_task/<int:task_id>', methods=['GET', 'POST'])
+def update_task(task_id):
+    task_manager = TaskManager.get_instance()
+    task = task_manager.get_task_by_id(task_id)
+    if not task:
+        return "Task not found", 404
+
+    if request.method == 'POST':
+        title = request.form['title']
+        description = request.form['description']
+        deadline = request.form['deadline']
+        priority = TaskPriority[request.form['priority']]
+        project = request.form['project']
+
+        task_manager.update_task(task_id, title, description, deadline, priority, project)
+        return redirect(url_for('.show_tasks'))
+
+    return render_template('task_form.html', task=task)
+
+
+@tasks_bp.route('/delete_task/<int:task_id>', methods=['POST'])
+def delete_task(task_id):
+    task_manager = TaskManager.get_instance()
+    task_manager.delete_task(task_id)
+    return redirect(url_for('.show_tasks'))
+
+
+@tasks_bp.route('/upload_csv', methods=['POST'])
+def upload_csv():
+    if 'csv_file' not in request.files:
+        return 'No file part', 400
+
+    csv_file = request.files['csv_file']
+
+    if csv_file.filename == '':
+        return 'No file selected', 400
+
+    if csv_file and csv_file.filename.endswith('.csv'):
+        try:
+            file_path = os.path.join(gettempdir(), csv_file.filename)
+            csv_file.save(file_path)
+
+            csv_manager = CSVManager()
+            csv_manager.import_tasks(file_path)
+            os.remove(file_path)
+        except Exception:
+            return 'Failed to import the CSV file, try uploading another one', 400
+
+        return redirect(url_for('.show_tasks'))
+
+    return 'Invalid file format, please upload a CSV file', 400
+
+
+@tasks_bp.route('/download_csv')
+def download_csv():
+    try:
+        csv_manager = CSVManager()
+        file_path = os.path.join(gettempdir(), 'tasks_export.csv')
+        csv_manager.export_tasks(file_path)
+    except Exception:
+        return 'Failed to export tasks to a CSV file', 400
+
+    @after_this_request
+    def remove_file(response):
+        try:
+            os.remove(file_path)
+        except OSError as err:
+            print(f'Failed to remove the exported CSV file: {err}', file=sys.stderr)
+        return response
+
+    return send_file(file_path, as_attachment=True, mimetype='text/csv')

projects/task_manager/static/list_style.css

@@ -0,0 +1,8 @@
+table {
+    table-layout: fixed;
+}
+
+td {
+    white-space: normal !important;
+    word-wrap: break-word;
+}