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