From 602dc04ccdaa442ac1dbf6b83f69c96a15bcca39 Mon Sep 17 00:00:00 2001 From: Mykhailo Date: Thu, 29 May 2025 23:26:52 +0200 Subject: [PATCH] Update solution.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit добавлен импорт Literal из typing добавлены в некоторые методы class TaskRepository типы StatusType, PriorityType и Order_byType, для большей информативности. get_tasks_by_status(self, status: StatusType) get_tasks_by_priority(self, priority: PriorityType) get_tasks_by_priority_range(self, min_priority: PriorityType, max_priority: PriorityType, order: Order_byType = 'ASC') get_all_tasks_sorted_by_priority(self, order: Order_byType = 'ASC', ascending: bool = True) get_tasks_by_status_sorted_by_priority(self, status: StatusType, order: Order_byType = 'ASC', ascending: bool = True) добавлена в несколько  методы возможность сортировки по ASC ||  DESC: def get_tasks_by_priority_range(self, min_priority: PriorityType, max_priority: PriorityType, order: Order_byType = 'ASC') def get_all_tasks_sorted_by_priority(self, order: Order_byType = 'ASC', ascending: bool = True) def get_tasks_by_status_sorted_by_priority(self, status: StatusType, order: Order_byType = 'ASC', ascending: bool = True) добавлено в метод def delete_completed_tasks(self) -> int вывод количества удаленных задач. Для большей информативности. добавлена новый метод def delete_all_tasks(self) -> int удаления всех задач с выводом количества всех удаленных задач. --- .../practice/TaskManager_part2/solution.py | 278 +++++++++++++----- 1 file changed, 197 insertions(+), 81 deletions(-) diff --git a/Lesson16/practice/TaskManager_part2/solution.py b/Lesson16/practice/TaskManager_part2/solution.py index e231b5a..9c1e54e 100644 --- a/Lesson16/practice/TaskManager_part2/solution.py +++ b/Lesson16/practice/TaskManager_part2/solution.py @@ -1,6 +1,7 @@ from connection import Connect from pathlib import Path -from typing import Optional +from typing import Optional, Literal + class Task: @@ -8,39 +9,55 @@ class Task: IN_PROGRESS = "In Progress" COMPLETED = "Completed" - def __init__(self, title, description="", status="Pending", priority=3, id=None): + def __init__(self, title, description="", status="Pending", priority=3, id = None): self.id = id self.title = title self.description = description self.status = status - if 1 <= priority <= 5: - self.priority = priority - else: - raise ValueError(...) + self.priority = priority + + def __repr__(self): + return f"Task(id={self.id}, title='{self.title}', description = '{self.description}', status = '{self.status}', priority={self.priority})" + # Добавьте методы для изменения статуса def mark_as_completed(self): self.status = Task.COMPLETED - def set_priority(self, new_priority): + def mark_as_in_progress(self): + self.status = Task.IN_PROGRESS + + def mark_as_pending(self): + self.status = Task.PENDING + + # Добавьте метод для изменения приоритета + def set_priority(self, new_priority: int): if 1 <= new_priority <= 5: self.priority = new_priority else: - raise ValueError(...) - - def __repr__(self): - return f"Task(id={self.id}, title='{self.title}', priority='{self.priority}')" + # print("new_priority должен быть в диапазоне от 1 до 5") + raise ValueError("new_priority должен быть в диапазоне от 1 до 5") -class TaskRepository: # TaskRepository -> TaskStorage +class TaskRepository: DB_FILE = Path("tasks.db") + # Типы + StatusType = Literal["Pending", "In Progress", "Completed"] + PriorityType = Literal[1, 2, 3, 4, 5] + Order_byType = Literal["ASC", "DESC"] def __init__(self): TaskRepository._ensure_db_table_exists() + @staticmethod + def _map_rows_to_tasks(tasks_data: list[tuple]) -> list[Task]: + tasks = [] + for data in tasks_data: + tasks.append(Task(*data[1:], data[0])) + return tasks + @classmethod def _ensure_db_table_exists(cls): """Создает таблицу tasks, если она еще не существует.""" - # cls.DB_FILE.parent.mkdir(parents=True, exist_ok=True) # Убедимся, что директория существует with Connect(cls.DB_FILE) as cursor: cursor.execute(''' CREATE TABLE IF NOT EXISTS tasks @@ -52,42 +69,37 @@ def _ensure_db_table_exists(cls): priority INTEGER CHECK ( priority BETWEEN 1 AND 5) DEFAULT 3 ); ''') - # commit() происходит автоматически при выходе из with Connect - def save(self, task: Task): """ Сохраняет или обновляет задачу в базе данных. Если id None, вставляет новую задачу. Иначе, обновляет существующую. """ - sql_insert = "INSERT INTO tasks (title, description, priority) VALUES (?, ?, ?)" - sql_update = """ + self._ensure_db_table_exists() + sql_insert = "INSERT INTO tasks (title, description, status, priority) VALUES (?, ?, ?, ?)" + sql_update = """ UPDATE tasks SET title = ?, description = ?, status = ?, priority = ? WHERE task_id = ?; """ if task.id: with Connect(TaskRepository.DB_FILE) as cursor: - cursor.execute(sql_update, (task.title, task.description, task.status, task.priority, self.id)) + cursor.execute(sql_update, (task.title, task.description, task.status, task.priority, task.id)) else: with Connect(TaskRepository.DB_FILE) as cursor: - cursor.execute(sql_insert, (task.title, task.description, task.priority)) + cursor.execute(sql_insert, (task.title, task.description, task.status, task.priority)) task.id = cursor.lastrowid - @staticmethod - def _map_rows_to_tasks(tasks_data: list[tuple]) -> list[Task]: - tasks = [] - for data in tasks_data: - tasks.append(Task(*data[1:], data[0])) - return tasks - def get_by_id(self, id) -> Optional['Task']: sql_select = "SELECT * FROM tasks WHERE task_id = ?" with Connect(self.DB_FILE) as cursor: cursor.execute(sql_select, (id,)) data = cursor.fetchone() - if data is None: + if data is not None: + # return Task(id=data[0], title=data[1], description=data[2], status=data[3], priority=data[4]) + return Task(*data[1:], data[0])# + else: + print(f"Задача с id={id} отсутствует") return None - return Task(*data[1:], data[0]) def get_all_tasks(self) -> list['Task']: sql_select = "SELECT * FROM tasks" @@ -95,70 +107,174 @@ def get_all_tasks(self) -> list['Task']: cursor.execute(sql_select) return self._map_rows_to_tasks(cursor.fetchall()) - def get_tasks_by_status(self, status: str) -> list[Task]: - sql = """ - SELECT * FROM tasks - WHERE status = ? - """ - with Connect(self.DB_FILE) as cursor: - cursor.execute(sql, (status, )) - return self._map_rows_to_tasks(cursor.fetchall()) - def get_tasks_by_priority_range(self, min_priority: int, max_priority: int) -> list[Task]: - sql = """ - SELECT * FROM tasks - WHERE priority BETWEEN ? AND ?; - """ - with Connect(self.DB_FILE) as cursor: - cursor.execute(sql, (min_priority, max_priority)) - return self._map_rows_to_tasks(cursor.fetchall()) def delete(self, task: Task): """Удаляет задачу из базы данных.""" - # TODO-3: реализуйте метод - if not task.id: - print(f"нельзя удалить задачу без id") - return sql_delete = "DELETE FROM tasks WHERE task_id = ?" - with Connect(TaskRepository.DB_FILE) as cursor: + if task is None or task.id is None: + print("Нельзя удалить: задача отсутствует или не имеет id") + return + with Connect(self.DB_FILE) as cursor: cursor.execute(sql_delete, (task.id,)) if cursor.rowcount > 0: print(f"Задача с id={task.id} удалена") - task.id = None else: print(f"Задача с id={task.id} не найдена") + task.id = None + + def get_tasks_by_status(self, status: StatusType) -> list[Task]: + # Возвращает список зада с определенным статусом. + sql_select = "SELECT * FROM tasks WHERE status = ?" + if status in ['Pending', 'In Progress', 'Completed']: + raise ValueError(f"status = {status} --> должен быть только из: 'Pending', 'In Progress', 'Completed'") + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql_select, (status,)) + return self._map_rows_to_tasks(cursor.fetchall()) + + def get_tasks_by_priority(self, priority: PriorityType) -> list[Task]: + # Возвращает список задач с определенным приоритетом. + sql_select = "SELECT * FROM tasks WHERE priority = ?" + if isinstance(priority, int) and 1 <= priority <= 5: + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql_select, (priority,)) + return self._map_rows_to_tasks(cursor.fetchall()) + else: + raise ValueError(f"priority --> должен быть integer или : 1 <= priority <= 5") + + def get_completed_tasks(self) -> list[Task]: + # Специализированный метод для получения всех завершенных задач. + sql_select = "SELECT * FROM tasks WHERE status = 'Completed'" + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql_select) + return self._map_rows_to_tasks(cursor.fetchall()) + + + def get_tasks_by_title_contains(self, keyword: str) -> list[Task]: + # Поиск задач по ключевому слову в заголовке. + if not isinstance(keyword, str): + raise TypeError("Ожидается строка для поиска в заголовке.") + if not keyword.strip(): + raise ValueError("Ключевое слово не должно быть пустым.") # если отправить пустую строку, выдаст всю базу, такая функция уже есть + sql_select = "SELECT * FROM tasks WHERE LOWER(title) LIKE LOWER(?)" + pattern = f"%{keyword}%" + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql_select, (pattern,)) + return self._map_rows_to_tasks(cursor.fetchall()) + def get_tasks_by_priority_range(self, min_priority: PriorityType, max_priority: PriorityType, order: Order_byType = 'ASC') -> list[Task]: + # Задачи в заданном диапазоне приоритетов. Методы получения задач с сортировкой: + if not all(isinstance(p, int) for p in (min_priority, max_priority)): + raise TypeError("Оба значения должны быть целыми числами.") + if not (1 <= min_priority <= 5 and 1 <= max_priority <= 5): + raise ValueError("Приоритет должен быть в диапазоне от 1 до 5.") + if min_priority > max_priority: + raise ValueError("Минимальный приоритет не может быть больше максимального.") + order = order.upper() + if order not in ['ASC', 'DESC']: + raise ValueError("Параметр 'order' должен быть 'ASC' или 'DESC'.") + sql_select = f""" + SELECT * FROM tasks + WHERE priority BETWEEN ? AND ? + ORDER BY priority {order} + """ + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql_select, (min_priority, max_priority)) + return self._map_rows_to_tasks(cursor.fetchall()) + + def get_all_tasks_sorted_by_priority(self, order: Order_byType = 'ASC', ascending: bool = True) -> list[Task]: + # Получает все задачи, отсортированные по приоритету. + order = order.upper() + if order not in ['ASC', 'DESC']: + raise ValueError("Параметр 'order' должен быть 'ASC' или 'DESC'.") + sql_select = f"SELECT * FROM tasks ORDER BY priority {order}" + if ascending: + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql_select) + return self._map_rows_to_tasks(cursor.fetchall()) + else: + return [] + + def get_tasks_by_status_sorted_by_priority(self, status: StatusType, order: Order_byType = 'ASC', ascending: bool = True) -> list[Task]: + #Получает задачи по статусу, отсортированные по приоритету. + if status not in ['Pending', 'In Progress', 'Completed']: + error_text = f"status = {status} --> должен быть только из: 'Pending', 'In Progress', 'Completed'" + raise ValueError(error_text) + order = order.upper() + if order not in ['ASC', 'DESC']: + raise ValueError("Параметр 'order' должен быть 'ASC' или 'DESC'.") + sql = f""" + SELECT * FROM tasks + WHERE status = ? ORDER BY priority {order} + """ + if ascending: + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql, (status,)) + return self._map_rows_to_tasks(cursor.fetchall()) + else: + return [] -# Использование -task_repository = TaskRepository() # Создаем экземпляр репозитория - -# Создаем новую задачу -# new_task = Task("Купить молоко", "Зайти в магазин", priority=1) -# task_repository.save(new_task) # Сохраняем новую задачу, new_task.id будет обновлен - -task = task_repository.get_by_id(id=1) -task.mark_as_completed() -if task: - task_repository.delete(task) -print(task) -# print(f"Сохранена новая задача: {new_task}") - -# Получаем задачу по ID -# retrieved_task = task_repository.get_by_id(new_task.id) -# if retrieved_task: -# print(f"Получена задача: {retrieved_task}") -# retrieved_task.description = "Купить 2 литра молока" -# retrieved_task.mark_as_in_progress() -# task_repository.save(retrieved_task) # Обновляем задачу -# print(f"Обновленная задача: {retrieved_task}") + def delete_completed_tasks(self) -> int: + # Удаляет все задачи со статусом "Завершено". + sql = f""" + DELETE FROM tasks + WHERE status = 'Completed' + """ + try: + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql) + deleted_count = cursor.rowcount # возвращает количество строк, затронутых последней операцией + print(f"удалено задач: {deleted_count}") + return deleted_count + except Exception as e: + raise Exception(f"Произошла непредвиденная ошибка при удалении завершенных задач: {e}") from e + + def delete_all_tasks(self) -> int: + # Удаляет все задачи со статусом "Завершено". + sql = f"DELETE FROM tasks" + try: + with Connect(self.DB_FILE) as cursor: + cursor.execute(sql) + deleted_count = cursor.rowcount # возвращает количество строк, затронутых последней операцией + # print(f"удалено задач: {deleted_count}") + return deleted_count + except Exception as e: + raise Exception(f"Произошла непредвиденная ошибка при удалении задач: {e}") from e + +################################### # -# # Получаем все задачи -# all_tasks = task_repository.get_all_tasks() -# print("\nВсе задачи:") -# for task in all_tasks: +################################### + +# tasks = [ +# Task("молоко в холодильнике", "найти молоко", status='Completed', priority=1), +# Task("написать отчет", "сделать отчет по проекту до понедельника", status='In Progress', priority=5), +# Task("купить хлеб", "зайти в ближайший магазин", status='Pending', priority=2), +# Task("помыть посуду", "особенно кастрюли", status='Completed', priority=3), +# Task("прочитать книгу", "прочитать хотя бы 50 страниц", status='In Progress', priority=4), +# Task("записаться к врачу", "позвонить в клинику и записаться", status='Pending', priority=5), +# Task("выучить 10 слов по немецкому", "использовать Quizlet", status='Pending', priority=4), +# Task("забрать посылку", "отделение почты работает до 18:00", status='Completed', priority=2), +# Task("позвонить другу", "спросить про планы на выходные", status='Pending', priority=3), +# Task("подготовить презентацию", "для встречи в пятницу", status='In Progress', priority=1), +# Task("оплатить интернет", "до 25 числа", status='Completed', priority=4), +# Task("найти новую статью", "для курсовой работы", status='In Progress', priority=5) +# ] +# # # # Использование +# task_repository = TaskRepository() # Создаем экземпляр репозитория +# for task in tasks: +# task_repository.save(task) + +# show_tasks = task_repository.get_all_tasks() +# show_tasks = task_repository.get_tasks_by_status("Completed") +# show_tasks = task_repository.get_tasks_by_priority(2+1+1) +# show_tasks = task_repository.get_completed_tasks() +# show_tasks = task_repository.get_tasks_by_priority_range(1,5, "DESC") +# show_tasks = task_repository.get_all_tasks_sorted_by_priority() +# show_tasks = task_repository.get_tasks_by_status_sorted_by_priority('Completed') +# show_tasks = task_repository.get_all_tasks() + +# for task in show_tasks: # print(task) -# -# # Удаляем задачу -# if retrieved_task: -# task_repository.delete(retrieved_task) -# print(task_repository.get_by_id(retrieved_task.id)) # Должно быть None + +# task_repository.delete_all_tasks() +# task_repository.delete_completed_tasks()