services/task_manager.py - master - fave/To_Do

services/task_manager.py
153 строки5 KB
fave
task_manager_main_ready
07 ноя 2025, 21:30
07 ноя 2025, 21:3043c9177
from datetime import datetime, date
from typing import List
from config.SQLite import SQLiteConfig
from models.task import Task
import sqlite3


class TaskManager:
    def __init__(self, db_path='todo_app.db'):
        self.config = SQLiteConfig(db_path)

    def add_task(self, task: Task) -> int:
        conn = self.config.get_connection()
        if conn is None:
            return None

        try:
            cursor = conn.cursor()
            due_date_str = task.due_date.isoformat() if task.due_date else None
            cursor.execute('''
                INSERT INTO tasks(title, description, due_date, is_completed) 
                VALUES(?, ?, ?, ?)
            ''', (task.title, task.description, due_date_str, task.is_completed))

            task_id = cursor.lastrowid
            conn.commit()
            print("✅ Задача была успешно добавлена")
            return task_id
        except sqlite3.Error as e:
            print(f"❌ Ошибка при добавлении задания: {e}")
            return None
        finally:
            conn.close()

    def get_all_tasks(self) -> List[Task]:
        conn = self.config.get_connection()
        if conn is None:
            return []

        try:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT id, title, description, created_date, due_date, is_completed
                FROM tasks
                ORDER BY created_date DESC
            ''')

            tasks = []
            for row in cursor:
                due_date = None
                if row[4]:  # due_date
                    try:
                        due_date = date.fromisoformat(row[4])
                    except:
                        due_date = None

                created_date = None
                if row[3]:  # created_date
                    try:
                        created_date = datetime.fromisoformat(row[3].replace(' ', 'T'))
                    except:
                        created_date = datetime.now()

                task = Task(
                    task_id=row[0],
                    title=row[1],
                    description=row[2],
                    due_date=due_date,
                    is_completed=bool(row[5]),
                    created_date=created_date
                )
                tasks.append(task)

            return tasks

        except sqlite3.Error as e:
            print(f"❌ Ошибка получения задач: {e}")
            return []
        finally:
            conn.close()


    def get_task_id(self, task_id:int)->Task:
        conn = self.config.get_connection()
        if conn is None:
            return None
        try:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM tasks WHERE id = ?',(task_id,))
            row = cursor.fetchone()
            if row:
                due_date = date.fromisoformat(row[4]) if row[4] else None
                return Task(
                    task_id = row[0],
                    title = row[1],
                    description=row[2],
                    due_date=due_date,
                    is_completed=bool(row[5])
                )
            return None
        except sqlite3.Error as e:
            print(f"Ошибка получения задания:{e}")
            return None
        finally:
            conn.close()


    def update_task(self, task:Task)->bool:
        conn = self.config.get_connection()
        if conn is None:
            return False

        try:
            cursor = conn.cursor()
            due_date = task.due_date.isoformat() if task.due_date else None
            cursor.execute('''
            UPDATE tasks
            SET title = ?, description = ?, due_date = ?, is_completed = ?
            WHERE id = ?
            ''',(task.title, task.description, task.due_date, task.is_completed, task.id))
            conn.commit()
            success = cursor.rowcount>0
            if success:
                print("Задача обновлена")
            else:
                print("Задача не найдена")
            return success
        except sqlite3.Error as e:
            print(f"Ошибка обновы:{e}")
            return False
        finally:
            conn.close()


    def delete_task(self, task:Task)->bool:
       conn = self.config.get_connection()
       if conn is None:
           return False
       try:
           cursor = conn.cursor()
           cursor.execute('DELETE FROM tasks WHERE id = ? ',(task.id,))
           conn.commit()
           success = cursor.rowcount>0
           if success:
               print("Запись удалена")
           else:
               print("Запись не найдена")
           return success
       except sqlite3.Error as e:
           print(f"Ошибка удаления:{e}")
           return False
       finally:
           conn.close()