Added relation_types to Task; code refactor

bode/bode/models/task.py

@@ -4,6 +4,7 @@
 from sqlalchemy import Enum
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.exc import IntegrityError, NoResultFound
+from sqlalchemy.ext.hybrid import hybrid_property
 
 from bode.app import db
 from bode.models.tag import Tag
@@ -35,6 +36,12 @@ class Task(db.Model):
 
     tags = db.relationship("Tag", secondary=task_tag, back_populates="task")
 
+    @hybrid_property
+    def relation_types(self):
+        from bode.models.task_relations_actions import get_relation_types
+
+        return get_relation_types(self.id)
+
     def get(task_id):
         return Task.query.get_or_404(task_id)
 

bode/bode/models/task_actions.py

@@ -5,11 +5,12 @@
 
 def delete_task(task_id):
     """Function deletes task, all it's relations and all it's subtasks reursively."""
+    from bode.models.task_relations_actions import get_related_tasks
 
     def is_subtask_relation(relation):
         return relation.type == RelationType.Subtask.value and str(relation.first_task_id) == task_id
 
-    relation_task_pairs = TaskRelation.get_related_tasks(task_id)
+    relation_task_pairs = get_related_tasks(task_id)
 
     for relation, related_task in relation_task_pairs:
         TaskRelation.delete(relation.id)
@@ -24,6 +25,7 @@ def is_subtask_relation(relation):
 
 def edit_task(task_id, **task_data):
     """Function edits task. If task is checked, all interchangable tasks with status todo will be indirectly checked."""
+    from bode.models.task_relations_actions import get_related_tasks
 
     def is_interchangable_relation(relation):
         return relation.type == RelationType.Interchangable.value and str(relation.first_task_id) == task_id
@@ -36,7 +38,7 @@ def is_interchangable_relation(relation):
     db.session.commit()
 
     if task_data["status"] != TaskStatus.TODO.value:
-        for relation, related_task in TaskRelation.get_related_tasks(task_id):
+        for relation, related_task in get_related_tasks(task_id):
             if is_interchangable_relation(relation):
                 if related_task.status != TaskStatus.TODO:
                     continue

bode/bode/models/task_relation.py

@@ -5,7 +5,6 @@
 from sqlalchemy.dialects.postgresql import ENUM, UUID
 
 from bode.app import db
-from bode.models.task import Task
 
 
 class RelationType(Enum):
@@ -82,27 +81,6 @@ def delete(relation_id):
 
         return relation
 
-    def get_lhs_related_tasks(task_id, filters=list()):
-        all_filters = [TaskRelation.first_task_id == task_id] + filters
-        return (
-            db.session.query(TaskRelation, Task)
-            .filter(*all_filters)
-            .join(Task, TaskRelation.second_task_id == Task.id)
-            .all()
-        )
-
-    def get_rhs_related_tasks(task_id, filters=list()):
-        all_filters = [TaskRelation.second_task_id == task_id] + filters
-        return (
-            db.session.query(TaskRelation, Task)
-            .filter(*all_filters)
-            .join(Task, TaskRelation.first_task_id == Task.id)
-            .all()
-        )
-
-    def get_related_tasks(task_id):
-        return TaskRelation.get_lhs_related_tasks(task_id) + TaskRelation.get_rhs_related_tasks(task_id)
-
     def __repr__(self):
         return f"""<TaskRelation
         {self.first_task_id} <{self.type}> {self.second_task_id}

bode/bode/models/task_relations_actions.py

@@ -0,0 +1,57 @@
+from bode.app import db
+from bode.models.task_relation import RelationType, TaskRelation
+from bode.resources.task_relations.schemas import DirectedRelationType
+
+
+def get_lhs_related_tasks(task_id, filters=list()):
+    from bode.models.task import Task
+
+    all_filters = [TaskRelation.first_task_id == task_id] + filters
+    return (
+        db.session.query(TaskRelation, Task)
+        .filter(*all_filters)
+        .join(Task, TaskRelation.second_task_id == Task.id)
+        .all()
+    )
+
+
+def get_rhs_related_tasks(task_id, filters=list()):
+    from bode.models.task import Task
+
+    all_filters = [TaskRelation.second_task_id == task_id] + filters
+    return (
+        db.session.query(TaskRelation, Task)
+        .filter(*all_filters)
+        .join(Task, TaskRelation.first_task_id == Task.id)
+        .all()
+    )
+
+
+def get_related_tasks(task_id):
+    return get_lhs_related_tasks(task_id) + get_rhs_related_tasks(task_id)
+
+
+def map_to_related_task_schema(relation: TaskRelation, task_id):
+    match relation.type:
+        case RelationType.Dependent.value:
+            type = (
+                DirectedRelationType.Blocks.value
+                if task_id == relation.first_task_id
+                else DirectedRelationType.IsBlockedBy.value
+            )
+        case RelationType.Subtask.value:
+            type = (
+                DirectedRelationType.Supertask.value
+                if task_id == relation.first_task_id
+                else DirectedRelationType.Subtask.value
+            )
+        case _:
+            type = DirectedRelationType.Interchangable.value
+
+    return type
+
+
+def get_relation_types(task_id):
+    filters = [TaskRelation.first_task_id == task_id or TaskRelation.second_task_id == task_id]
+    relations = TaskRelation.query.filter(*filters).all()
+    return {map_to_related_task_schema(relation, task_id) for relation in relations}

bode/bode/resources/task_relations/api.py

@@ -6,6 +6,10 @@
 
 from bode.models.task import Task
 from bode.models.task_relation import RelationType, TaskRelation
+from bode.models.task_relations_actions import (
+    get_lhs_related_tasks,
+    get_rhs_related_tasks,
+)
 from bode.resources.task_relations.schemas import (
     LHS_RELATION_TYPES,
     RHS_RELATION_TYPES,
@@ -93,15 +97,13 @@ def get(self, params: DirectedRelationSchema, task_id: UUID):
         query_result = []
 
         if not relation_type:
-            query_result += TaskRelation.get_lhs_related_tasks(task_id)
-            query_result += TaskRelation.get_rhs_related_tasks(
-                task_id, [TaskRelation.type != RelationType.Interchangable.value]
-            )
+            query_result += get_lhs_related_tasks(task_id)
+            query_result += get_rhs_related_tasks(task_id, [TaskRelation.type != RelationType.Interchangable.value])
 
         elif relation_type in SYMMETRIC_RELATION_TYPES + LHS_RELATION_TYPES:
-            query_result += TaskRelation.get_lhs_related_tasks(task_id, [self.lhs_relation_filter(relation_type)])
+            query_result += get_lhs_related_tasks(task_id, [self.lhs_relation_filter(relation_type)])
 
         elif relation_type in RHS_RELATION_TYPES:
-            query_result += TaskRelation.get_rhs_related_tasks(task_id, [self.rhs_relation_filter(relation_type)])
+            query_result += get_rhs_related_tasks(task_id, [self.rhs_relation_filter(relation_type)])
 
         return map(lambda pair: self.map_to_related_task_schema(*pair), query_result)

bode/bode/resources/tasks/schemas.py

@@ -23,3 +23,4 @@ class TaskSchema(BaseSchema):
     due_date = fields.DateTime()
     status = fields.String()
     tags = fields.List(fields.Nested(TagSchema))
+    relation_types = fields.List(fields.String())