projectal.entities.task

  1import datetime
  2import copy
  3import sys
  4import projectal
  5from projectal.entity import Entity
  6from projectal.enums import DateLimit
  7from projectal.linkers import *
  8
  9
 10class Task(
 11    Entity,
 12    ResourceLinker,
 13    SkillLinker,
 14    FileLinker,
 15    StageLinker,
 16    StaffLinker,
 17    RebateLinker,
 18    NoteLinker,
 19    TagLinker,
 20    PredecessorTaskLinker,
 21    TimelogLinker,
 22):
 23    """
 24    Implementation of the [Task](https://projectal.com/docs/latest/#tag/Task) API.
 25    """
 26
 27    _path = "task"
 28    _name = "task"
 29    _links = [
 30        ResourceLinker,
 31        SkillLinker,
 32        FileLinker,
 33        StageLinker,
 34        StaffLinker,
 35        RebateLinker,
 36        NoteLinker,
 37        TagLinker,
 38        TimelogLinker,
 39    ]
 40    _links_reverse = [
 41        PredecessorTaskLinker,
 42    ]
 43
 44    def _add_link_def(self, cls, reverse=False):
 45        """
 46        Each entity is accompanied by a dict with details about how to
 47        get access to the data of the link within the object. Subclasses
 48        can pass in customizations to this dict when their APIs differ.
 49
 50        reverse denotes a reverse linker, where extra work is done to
 51        reverse the relationship of the link internally so that it works.
 52        The backend only offers one side of the relationship.
 53        """
 54        d = {
 55            "name": cls._link_name,
 56            "link_key": cls._link_key or cls._link_name + "List",
 57            "data_name": cls._link_data_name,
 58            "type": cls._link_type,
 59            "entity": cls._link_entity or cls._link_name.capitalize(),
 60            "reverse": reverse,
 61        }
 62        self._link_def_by_key[d["link_key"]] = d
 63        self._link_def_by_name[d["name"]] = d
 64        if cls._link_name == "predecessor_task":
 65            d_after_reverse = copy.deepcopy(d)
 66            d_after_reverse["reverse"] = False
 67            self._link_def_by_name["task"] = d_after_reverse
 68            # We need this to be present in the link def so that
 69            # returned predecessor tasks can be typed as Tasks
 70            d_for_pred_link_typing = copy.deepcopy(d)
 71            d_for_pred_link_typing["link_key"] = "planList"
 72            self._link_def_by_key[
 73                d_for_pred_link_typing["link_key"]
 74            ] = d_for_pred_link_typing
 75
 76    @classmethod
 77    def create(
 78        cls,
 79        holder,
 80        entities,
 81        batch_linking=True,
 82        disable_system_features=True,
 83        enable_system_features_on_exit=True,
 84    ):
 85        """Create a Task
 86
 87        `holder`: An instance or the `uuId` of the owner
 88
 89        `entities`: `dict` containing the fields of the entity to be created,
 90        or a list of such `dict`s to create in bulk.
 91        """
 92        holder_id = holder["uuId"] if isinstance(holder, dict) else holder
 93        params = "?holder=" + holder_id
 94        out = super().create(
 95            entities,
 96            params,
 97            batch_linking,
 98            disable_system_features,
 99            enable_system_features_on_exit,
100        )
101
102        # Tasks should always refer to their parent and project. We don't get this information
103        # from the creation api method, but we can insert them ourselves because we know what
104        # they are.
105        def add_fields(obj):
106            obj.set_readonly("projectRef", holder_id)
107            obj.set_readonly("parent", obj.get("parent", holder_id))
108
109        if isinstance(out, dict):
110            add_fields(out)
111        if isinstance(out, list):
112            for obj in out:
113                add_fields(obj)
114        return out
115
116    @classmethod
117    def get(cls, entities, links=None, deleted_at=None):
118        r = super().get(entities, links, deleted_at)
119        if not links:
120            return r
121        # When Predecessor Task links are fetched,
122        # make sure the key matches the name expected
123        # by the predecessor linking REST end point
124        if PredecessorTaskLinker._link_name.casefold() in (
125            link.casefold() for link in links
126        ):
127            if isinstance(r, dict):
128                r["taskList"] = r.pop("planList", [])
129                r._Entity__old = copy.deepcopy(r)
130            else:
131                for entity in r:
132                    entity["taskList"] = entity.pop("planList", [])
133                    entity._Entity__old = copy.deepcopy(entity)
134        return r
135
136    # Override here to correctly format the URL for the Predecessor Task link case
137    def _link(
138        self, to_entity_name, to_link, operation, update_cache=True, batch_linking=True
139    ):
140        """
141        `to_entity_name`: Destination entity name (e.g. 'staff')
142
143        `to_link`: List of Entities of the same type (and optional data) to link to
144
145        `operation`: `add`, `update`, `delete`
146
147        'update_cache': also modify the entity's internal representation of the links
148        to match the operation that was done. Set this to False when replacing the
149        list with a new one (i.e., when calling save() instead of a linker method).
150
151        'batch_linking': Enabled by default, batches any link
152        updates required into composite API requests. If disabled
153        a request will be executed for each link update.
154        Recommended to leave enabled to increase performance.
155        """
156
157        link_def = self._link_def_by_name[to_entity_name]
158        to_key = link_def["link_key"]
159
160        if isinstance(to_link, dict) and link_def["type"] == list:
161            # Convert input dict to list when link type is a list (we allow linking to single entity for convenience)
162            to_link = [to_link]
163
164            # For cases where user passed in dict instead of Entity, we turn them into
165            # Entity on their behalf.
166            typed_list = []
167            target_cls = getattr(sys.modules["projectal.entities"], link_def["entity"])
168            for link in to_link:
169                if not isinstance(link, target_cls):
170                    typed_list.append(target_cls(link))
171                else:
172                    typed_list.append(link)
173            to_link = typed_list
174        else:
175            # For everything else, we expect types to match.
176            if not isinstance(to_link, link_def["type"]):
177                raise api.UsageException(
178                    "Expected link type to be {}. Got {}.".format(
179                        link_def["type"], type(to_link)
180                    )
181                )
182
183        if not to_link:
184            return
185
186        url = ""
187        payload = {}
188        request_list = []
189        # Is it a reverse linker? If so, invert the relationship
190        if link_def["reverse"]:
191            for link in to_link:
192                # Sets the data attribute on the correct
193                # link entity
194                if link_def["name"] == "predecessor_task":
195                    data_name = link_def.get("data_name")
196                    self[data_name] = copy.deepcopy(link[data_name])
197                request_list.extend(
198                    link._link(
199                        self._name,
200                        self,
201                        operation,
202                        update_cache,
203                        batch_linking=batch_linking,
204                    )
205                )
206        else:
207            # Only keep UUID and the data attribute, if it has one
208            def strip_payload(link):
209                single = {"uuId": link["uuId"]}
210                data_name = link_def.get("data_name")
211                if data_name and data_name in link:
212                    single[data_name] = copy.deepcopy(link[data_name])
213                    # limiting data attribute removal to only planLink
214                    # in case of side effects
215                    if data_name == "planLink":
216                        del link[data_name]
217                return single
218
219            # If batch linking is enabled and the entity to link is a list of entities,
220            # a separate request must be constructed for each one because the final composite
221            # request permits only one input per call
222            if to_entity_name == "predecessor_task" or to_entity_name == "task":
223                url = "/api/{}/plan/TASK/{}".format(self._path, operation)
224            else:
225                url = "/api/{}/link/{}/{}".format(self._path, to_entity_name, operation)
226            to_link_payload = None
227            if isinstance(to_link, list):
228                to_link_payload = []
229                for link in to_link:
230                    if batch_linking:
231                        request_list.append(
232                            {
233                                "method": "POST",
234                                "invoke": url,
235                                "body": {
236                                    "uuId": self["uuId"],
237                                    to_key: [strip_payload(link)],
238                                },
239                            }
240                        )
241                    else:
242                        to_link_payload.append(strip_payload(link))
243            if isinstance(to_link, dict):
244                if batch_linking:
245                    request_list.append(
246                        {
247                            "method": "POST",
248                            "invoke": url,
249                            "body": {
250                                "uuId": self["uuId"],
251                                to_key: strip_payload(to_link),
252                            },
253                        }
254                    )
255                else:
256                    to_link_payload = strip_payload(to_link)
257
258            if not batch_linking:
259                payload = {"uuId": self["uuId"], to_key: to_link_payload}
260                api.post(url, payload=payload)
261
262        if not update_cache:
263            return request_list
264
265        # Set the initial state if first add. We need the type to be set to correctly update the cache
266        if operation == "add" and self.get(to_key, None) is None:
267            if link_def.get("type") == dict:
268                self[to_key] = {}
269            elif link_def.get("type") == list:
270                self[to_key] = []
271
272        # Modify the entity object's cache of links to match the changes we pushed to the server.
273        if isinstance(self.get(to_key, []), list):
274            if operation == "add":
275                # Sometimes the backend doesn't return a list when it has none. Create it.
276                if to_key not in self:
277                    self[to_key] = []
278
279                for to_entity in to_link:
280                    self[to_key].append(to_entity)
281            else:
282                for to_entity in to_link:
283                    # Find it in original list
284                    for i, old in enumerate(self.get(to_key, [])):
285                        if old["uuId"] == to_entity["uuId"]:
286                            if operation == "update":
287                                self[to_key][i] = to_entity
288                            elif operation == "delete":
289                                del self[to_key][i]
290        if isinstance(self.get(to_key, None), dict):
291            if operation in ["add", "update"]:
292                self[to_key] = to_link
293            elif operation == "delete":
294                self[to_key] = None
295
296        # Update the "old" record of the link on the entity to avoid
297        # flagging it for changes (link lists are not meant to be user editable).
298        if to_key in self:
299            self._Entity__old[to_key] = self[to_key]
300
301        return request_list
302
303    def update_order(self, order_at_uuId, order_as=True):
304        url = "/api/task/update?order-at={}&order-as={}".format(
305            order_at_uuId, "true" if order_as else "false"
306        )
307        return api.put(url, [{"uuId": self["uuId"]}])
308
309    def link_predecessor_task(self, predecessor_task):
310        return self.__plan(self, predecessor_task, "add")
311
312    def relink_predecessor_task(self, predecessor_task):
313        return self.__plan(self, predecessor_task, "update")
314
315    def unlink_predecessor_task(self, predecessor_task):
316        return self.__plan(self, predecessor_task, "delete")
317
318    @classmethod
319    def __plan(cls, from_task, to_task, operation):
320        url = "/api/task/plan/task/{}".format(operation)
321        # Invert the link relationship to match the linker
322        if isinstance(to_task, dict):
323            from_task_copy = copy.deepcopy(from_task)
324            from_task_copy[PredecessorTaskLinker._link_data_name] = copy.deepcopy(
325                to_task[PredecessorTaskLinker._link_data_name]
326            )
327            payload = {"uuId": to_task["uuId"], "taskList": [from_task_copy]}
328            api.post(url, payload=payload)
329        elif isinstance(to_task, list):
330            for task in to_task:
331                from_task_copy = copy.deepcopy(from_task)
332                from_task_copy[PredecessorTaskLinker._link_data_name] = copy.deepcopy(
333                    task[PredecessorTaskLinker._link_data_name]
334                )
335                payload = {"uuId": task["uuId"], "taskList": [from_task_copy]}
336                api.post(url, payload=payload)
337        return True
338
339    def parents(self):
340        """
341        Return an ordered list of [name, uuId] pairs of this task's parents, up to
342        (but not including) the root of the project.
343        """
344        payload = {
345            "name": "Task Parents",
346            "type": "msql",
347            "start": 0,
348            "limit": -1,
349            "holder": "{}".format(self["uuId"]),
350            "select": [
351                ["TASK(one).PARENT_ALL_TASK.name"],
352                ["TASK(one).PARENT_ALL_TASK.uuId"],
353            ],
354        }
355        list = api.query(payload)
356        # Results come back in reverse order. Flip them around
357        list.reverse()
358        return list
359
360    def project_uuId(self):
361        """Return the `uuId` of the Project that holds this Task."""
362        payload = {
363            "name": "Project that holds this task",
364            "type": "msql",
365            "start": 0,
366            "limit": 1,
367            "holder": "{}".format(self["uuId"]),
368            "select": [["TASK.PROJECT.uuId"]],
369        }
370        projects = api.query(payload)
371        for t in projects:
372            return t[0]
373        return None
374
375    @classmethod
376    def add_task_template(cls, project, template):
377        """Insert TaskTemplate `template` into Project `project`"""
378        url = "/api/task/task_template/add?override=false&group=false"
379        payload = {"uuId": project["uuId"], "templateList": [template]}
380        api.post(url, payload)
381
382    def reset_duration(self, calendars=None):
383        """Set this task's duration based on its start and end dates while
384        taking into account the calendar for weekends and scheduled time off.
385
386        calendars is expected to be the list of calendar objects for the
387        location of the project that holds this task. You may provide this
388        list yourself for efficiency (recommended) - if not provided, it
389        will be fetched for you by issuing requests to the server.
390        """
391        if not calendars:
392            if "projectRef" not in self:
393                task = projectal.Task.get(self)
394                project_ref = task["projectRef"]
395            else:
396                project_ref = self["projectRef"]
397            project = projectal.Project.get(project_ref, links=["LOCATION"])
398            for location in project.get("locationList", []):
399                calendars = location.calendar()
400                break
401
402        start = self.get("startTime")
403        end = self.get("closeTime")
404        if not start or start == DateLimit.Min:
405            return 0
406        if not end or end == DateLimit.Max:
407            return 0
408
409        # Build a list of weekday names that are non-working
410        base_non_working = set()
411        location_non_working = {}
412        location_working = set()
413        for calendar in calendars:
414            if calendar["name"] == "base_calendar":
415                for item in calendar["calendarList"]:
416                    if not item["isWorking"]:
417                        base_non_working.add(item["type"])
418
419            if calendar["name"] == "location":
420                for item in calendar["calendarList"]:
421                    start_date = datetime.date.fromisoformat(item["startDate"])
422                    end_date = datetime.date.fromisoformat(item["endDate"])
423                    if not item["isWorking"]:
424                        delta = start_date - end_date
425                        location_non_working[item["startDate"]] = delta.days + 1
426                    else:
427                        location_working = {
428                            (start_date + datetime.timedelta(days=x)).strftime(
429                                "%Y-%m-%d"
430                            )
431                            for x in range((end_date - start_date).days + 1)
432                        }
433
434        start = datetime.datetime.fromtimestamp(start / 1000)
435        end = datetime.datetime.fromtimestamp(end / 1000)
436        minutes = 0
437        current = start
438        while current <= end:
439            if (
440                current.strftime("%A") in base_non_working
441                and current.strftime("%Y-%m-%d") not in location_working
442            ):
443                current += datetime.timedelta(days=1)
444                continue
445            if current.strftime("%Y-%m-%d") in location_non_working:
446                days = location_non_working[current.strftime("%Y-%m-%d")]
447                current += datetime.timedelta(days=days)
448                continue
449            minutes += 8 * 60
450            current += datetime.timedelta(days=1)
451
452        self["duration"] = minutes