projectal.entities.task

  1import datetime
  2
  3import projectal
  4from projectal.entity import Entity
  5from projectal.enums import DateLimit
  6from projectal.linkers import *
  7
  8
  9class Task(
 10    Entity,
 11    ResourceLinker,
 12    SkillLinker,
 13    FileLinker,
 14    StageLinker,
 15    StaffLinker,
 16    RebateLinker,
 17    NoteLinker,
 18    TagLinker,
 19):
 20    """
 21    Implementation of the [Task](https://projectal.com/docs/latest/#tag/Task) API.
 22    """
 23
 24    _path = "task"
 25    _name = "task"
 26    _links = [
 27        ResourceLinker,
 28        SkillLinker,
 29        FileLinker,
 30        StageLinker,
 31        StaffLinker,
 32        RebateLinker,
 33        NoteLinker,
 34        TagLinker,
 35    ]
 36
 37    @classmethod
 38    def create(cls, holder, entities):
 39        """Create a Task
 40
 41        `holder`: An instance or the `uuId` of the owner
 42
 43        `entities`: `dict` containing the fields of the entity to be created,
 44        or a list of such `dict`s to create in bulk.
 45        """
 46        holder_id = holder["uuId"] if isinstance(holder, dict) else holder
 47        params = "?holder=" + holder_id
 48        out = super().create(entities, params)
 49
 50        # Tasks should always refer to their parent and project. We don't get this information
 51        # from the creation api method, but we can insert them ourselves because we know what
 52        # they are.
 53        def add_fields(obj):
 54            obj.set_readonly("projectRef", holder_id)
 55            obj.set_readonly("parent", obj.get("parent", holder_id))
 56
 57        if isinstance(out, dict):
 58            add_fields(out)
 59        if isinstance(out, list):
 60            for obj in out:
 61                add_fields(obj)
 62        return out
 63
 64    def update_order(self, order_at_uuId, order_as=True):
 65        url = "/api/task/update?order-at={}&order-as={}".format(
 66            order_at_uuId, "true" if order_as else "false"
 67        )
 68        return api.put(url, [{"uuId": self["uuId"]}])
 69
 70    def link_predecessor_task(self, predecessor_task):
 71        return self.__plan(self, predecessor_task, "add")
 72
 73    def relink_predecessor_task(self, predecessor_task):
 74        return self.__plan(self, predecessor_task, "update")
 75
 76    def unlink_predecessor_task(self, predecessor_task):
 77        return self.__plan(self, predecessor_task, "delete")
 78
 79    @classmethod
 80    def __plan(cls, from_task, to_task, operation):
 81        url = "/api/task/plan/task/{}".format(operation)
 82        payload = {"uuId": from_task["uuId"], "taskList": [to_task]}
 83        api.post(url, payload=payload)
 84        return True
 85
 86    def parents(self):
 87        """
 88        Return an ordered list of [name, uuId] pairs of this task's parents, up to
 89        (but not including) the root of the project.
 90        """
 91        payload = {
 92            "name": "Task Parents",
 93            "type": "msql",
 94            "start": 0,
 95            "limit": -1,
 96            "holder": "{}".format(self["uuId"]),
 97            "select": [
 98                ["TASK(one).PARENT_ALL_TASK.name"],
 99                ["TASK(one).PARENT_ALL_TASK.uuId"],
100            ],
101        }
102        list = api.query(payload)
103        # Results come back in reverse order. Flip them around
104        list.reverse()
105        return list
106
107    def project_uuId(self):
108        """Return the `uuId` of the Project that holds this Task."""
109        payload = {
110            "name": "Project that holds this task",
111            "type": "msql",
112            "start": 0,
113            "limit": 1,
114            "holder": "{}".format(self["uuId"]),
115            "select": [["TASK.PROJECT.uuId"]],
116        }
117        projects = api.query(payload)
118        for t in projects:
119            return t[0]
120        return None
121
122    @classmethod
123    def add_task_template(cls, project, template):
124        """Insert TaskTemplate `template` into Project `project`"""
125        url = "/api/task/task_template/add?override=false&group=false"
126        payload = {"uuId": project["uuId"], "templateList": [template]}
127        api.post(url, payload)
128
129    def reset_duration(self, calendars=None):
130        """Set this task's duration based on its start and end dates while
131        taking into account the calendar for weekends and scheduled time off.
132
133        calendars is expected to be the list of calendar objects for the
134        location of the project that holds this task. You may provide this
135        list yourself for efficiency (recommended) - if not provided, it
136        will be fetched for you by issuing requests to the server.
137        """
138        if not calendars:
139            if "projectRef" not in self:
140                task = projectal.Task.get(self)
141                project_ref = task["projectRef"]
142            else:
143                project_ref = self["projectRef"]
144            project = projectal.Project.get(project_ref, links=["LOCATION"])
145            for location in project.get("locationList", []):
146                calendars = location.calendar()
147                break
148
149        start = self.get("startTime")
150        end = self.get("closeTime")
151        if not start or start == DateLimit.Min:
152            return 0
153        if not end or end == DateLimit.Max:
154            return 0
155
156        # Build a list of weekday names that are non-working
157        base_non_working = set()
158        location_non_working = {}
159        location_working = set()
160        for calendar in calendars:
161            if calendar["name"] == "base_calendar":
162                for item in calendar["calendarList"]:
163                    if not item["isWorking"]:
164                        base_non_working.add(item["type"])
165
166            if calendar["name"] == "location":
167                for item in calendar["calendarList"]:
168                    start_date = datetime.date.fromisoformat(item["startDate"])
169                    end_date = datetime.date.fromisoformat(item["endDate"])
170                    if not item["isWorking"]:
171                        delta = start_date - end_date
172                        location_non_working[item["startDate"]] = delta.days + 1
173                    else:
174                        location_working = {
175                            (start_date + datetime.timedelta(days=x)).strftime(
176                                "%Y-%m-%d"
177                            )
178                            for x in range((end_date - start_date).days + 1)
179                        }
180
181        start = datetime.datetime.fromtimestamp(start / 1000)
182        end = datetime.datetime.fromtimestamp(end / 1000)
183        minutes = 0
184        current = start
185        while current <= end:
186            if (
187                current.strftime("%A") in base_non_working
188                and current.strftime("%Y-%m-%d") not in location_working
189            ):
190                current += datetime.timedelta(days=1)
191                continue
192            if current.strftime("%Y-%m-%d") in location_non_working:
193                days = location_non_working[current.strftime("%Y-%m-%d")]
194                current += datetime.timedelta(days=days)
195                continue
196            minutes += 8 * 60
197            current += datetime.timedelta(days=1)
198
199        self["duration"] = minutes