forked from ashishps1/awesome-low-level-design
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtask_manager.py
77 lines (65 loc) · 2.72 KB
/
task_manager.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
from datetime import datetime
from task import Task, TaskStatus
from user import User
class TaskManager:
_instance = None
def __init__(self):
if TaskManager._instance is not None:
raise Exception("This class is a singleton!")
else:
TaskManager._instance = self
self.tasks = {}
self.user_tasks = {}
@staticmethod
def get_instance():
if TaskManager._instance is None:
TaskManager()
return TaskManager._instance
def create_task(self, task):
self.tasks[task.get_id()] = task
self._assign_task_to_user(task.get_assigned_user(), task)
def update_task(self, updated_task):
existing_task = self.tasks.get(updated_task.get_id())
if existing_task:
existing_task.set_title(updated_task.get_title())
existing_task.set_description(updated_task.get_description())
existing_task.set_due_date(updated_task.get_due_date())
existing_task.set_priority(updated_task.get_priority())
existing_task.set_status(updated_task.get_status())
previous_user = existing_task.get_assigned_user()
new_user = updated_task.get_assigned_user()
if previous_user != new_user:
self._unassign_task_from_user(previous_user, existing_task)
self._assign_task_to_user(new_user, existing_task)
def delete_task(self, task_id):
task = self.tasks.pop(task_id, None)
if task:
self._unassign_task_from_user(task.get_assigned_user(), task)
def search_tasks(self, keyword):
matching_tasks = []
for task in self.tasks.values():
if keyword in task.get_title() or keyword in task.get_description():
matching_tasks.append(task)
return matching_tasks
def filter_tasks(self, status, start_date, end_date, priority):
filtered_tasks = []
for task in self.tasks.values():
if (
task.get_status() == status
and start_date <= task.get_due_date() <= end_date
and task.get_priority() == priority
):
filtered_tasks.append(task)
return filtered_tasks
def mark_task_as_completed(self, task_id):
task = self.tasks.get(task_id)
if task:
task.set_status(TaskStatus.COMPLETED)
def get_task_history(self, user):
return self.user_tasks.get(user.get_id(), [])
def _assign_task_to_user(self, user, task):
self.user_tasks.setdefault(user.get_id(), []).append(task)
def _unassign_task_from_user(self, user, task):
tasks = self.user_tasks.get(user.get_id())
if tasks:
tasks.remove(task)