import logging
from datetime import date, datetime
from dateutil.parser import parse
from .utils import NEXT_LINK_KEYWORD, ApiComponent, Pagination
log = logging.getLogger(__name__)
[docs]
class TaskDetails(ApiComponent):
_endpoints = {'task_detail': '/planner/tasks/{id}/details'}
[docs]
def __init__(self, *, parent=None, con=None, **kwargs):
""" A Microsoft O365 plan details
:param parent: parent object
:type parent: Task
:param Connection con: connection to use if no parent specified
:param Protocol protocol: protocol to use if no parent specified
(kwargs)
:param str main_resource: use this resource instead of parent resource
(kwargs)
"""
if parent and con:
raise ValueError('Need a parent or a connection but not both')
self.con = parent.con if parent else con
cloud_data = kwargs.get(self._cloud_data_key, {})
self.object_id = cloud_data.get('id')
# Choose the main_resource passed in kwargs over parent main_resource
main_resource = kwargs.pop('main_resource', None) or (
getattr(parent, 'main_resource', None) if parent else None)
main_resource = '{}{}'.format(main_resource, '')
super().__init__(
protocol=parent.protocol if parent else kwargs.get('protocol'),
main_resource=main_resource)
self.description = cloud_data.get(self._cc('description'), '')
self.references = cloud_data.get(self._cc('references'), '')
self.checklist = cloud_data.get(self._cc('checklist'), '')
self.preview_type = cloud_data.get(self._cc('previewType'), '')
self._etag = cloud_data.get('@odata.etag', '')
def __str__(self):
return self.__repr__()
def __repr__(self):
return 'Task Details'
def __eq__(self, other):
return self.object_id == other.object_id
[docs]
def update(self, **kwargs):
"""Updates this task detail
:param kwargs: all the properties to be updated.
:param dict checklist: the collection of checklist items on the task.
.. code-block::
e.g. checklist = {
"string GUID": {
"isChecked": bool,
"orderHint": string,
"title": string
}
} (kwargs)
:param str description: description of the task
:param str preview_type: this sets the type of preview that shows up on the task.
The possible values are: automatic, noPreview, checklist, description, reference.
:param dict references: the collection of references on the task.
.. code-block::
e.g. references = {
"URL of the resource" : {
"alias": string,
"previewPriority": string, #same as orderHint
"type": string, #e.g. PowerPoint, Excel, Word, Pdf...
}
}
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
_unsafe = ".:@#"
url = self.build_url(
self._endpoints.get("task_detail").format(id=self.object_id)
)
data = {
self._cc(key): value
for key, value in kwargs.items()
if key
in (
"checklist",
"description",
"preview_type",
"references",
)
}
if not data:
return False
if "references" in data and isinstance(data["references"], dict):
for key in list(data["references"].keys()):
if (
isinstance(data["references"][key], dict)
and not "@odata.type" in data["references"][key]
):
data["references"][key]["@odata.type"] = (
"#microsoft.graph.plannerExternalReference"
)
if any(u in key for u in _unsafe):
sanitized_key = "".join(
[
chr(b)
if b not in _unsafe.encode("utf-8", "strict")
else "%{:02X}".format(b)
for b in key.encode("utf-8", "strict")
]
)
data["references"][sanitized_key] = data["references"].pop(key)
if "checklist" in data:
for key in data["checklist"].keys():
if (
isinstance(data["checklist"][key], dict)
and not "@odata.type" in data["checklist"][key]
):
data["checklist"][key]["@odata.type"] = (
"#microsoft.graph.plannerChecklistItem"
)
response = self.con.patch(
url,
data=data,
headers={"If-Match": self._etag, "Prefer": "return=representation"},
)
if not response:
return False
new_data = response.json()
for key in data:
value = new_data.get(key, None)
if value is not None:
setattr(self, self.protocol.to_api_case(key), value)
self._etag = new_data.get("@odata.etag")
return True
[docs]
class PlanDetails(ApiComponent):
_endpoints = {"plan_detail": "/planner/plans/{id}/details"}
[docs]
def __init__(self, *, parent=None, con=None, **kwargs):
"""A Microsoft O365 plan details
:param parent: parent object
:type parent: Plan
:param Connection con: connection to use if no parent specified
:param Protocol protocol: protocol to use if no parent specified
(kwargs)
:param str main_resource: use this resource instead of parent resource
(kwargs)
"""
if parent and con:
raise ValueError("Need a parent or a connection but not both")
self.con = parent.con if parent else con
cloud_data = kwargs.get(self._cloud_data_key, {})
self.object_id = cloud_data.get("id")
# Choose the main_resource passed in kwargs over parent main_resource
main_resource = kwargs.pop("main_resource", None) or (
getattr(parent, "main_resource", None) if parent else None
)
main_resource = "{}{}".format(main_resource, "")
super().__init__(
protocol=parent.protocol if parent else kwargs.get("protocol"),
main_resource=main_resource,
)
self.shared_with = cloud_data.get(self._cc("sharedWith"), "")
self.category_descriptions = cloud_data.get(
self._cc("categoryDescriptions"), ""
)
self._etag = cloud_data.get("@odata.etag", "")
def __str__(self):
return self.__repr__()
def __repr__(self):
return "Plan Details"
def __eq__(self, other):
return self.object_id == other.object_id
[docs]
def update(self, **kwargs):
"""Updates this plan detail
:param kwargs: all the properties to be updated.
:param dict shared_with: dict where keys are user_ids and values are boolean (kwargs)
:param dict category_descriptions: dict where keys are category1, category2, ..., category25 and values are the label associated with (kwargs)
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
url = self.build_url(
self._endpoints.get("plan_detail").format(id=self.object_id)
)
data = {
self._cc(key): value
for key, value in kwargs.items()
if key in ("shared_with", "category_descriptions")
}
if not data:
return False
response = self.con.patch(
url,
data=data,
headers={"If-Match": self._etag, "Prefer": "return=representation"},
)
if not response:
return False
new_data = response.json()
for key in data:
value = new_data.get(key, None)
if value is not None:
setattr(self, self.protocol.to_api_case(key), value)
self._etag = new_data.get("@odata.etag")
return True
[docs]
class Task(ApiComponent):
"""A Microsoft Planner task"""
_endpoints = {
"get_details": "/planner/tasks/{id}/details",
"task": "/planner/tasks/{id}",
}
task_details_constructor = TaskDetails
[docs]
def __init__(self, *, parent=None, con=None, **kwargs):
"""A Microsoft planner task
:param parent: parent object
:type parent: Planner or Plan or Bucket
:param Connection con: connection to use if no parent specified
:param Protocol protocol: protocol to use if no parent specified
(kwargs)
:param str main_resource: use this resource instead of parent resource
(kwargs)
"""
if parent and con:
raise ValueError("Need a parent or a connection but not both")
self.con = parent.con if parent else con
cloud_data = kwargs.get(self._cloud_data_key, {})
self.object_id = cloud_data.get("id")
# Choose the main_resource passed in kwargs over parent main_resource
main_resource = kwargs.pop("main_resource", None) or (
getattr(parent, "main_resource", None) if parent else None
)
main_resource = "{}{}".format(main_resource, "")
super().__init__(
protocol=parent.protocol if parent else kwargs.get("protocol"),
main_resource=main_resource,
)
self.plan_id = cloud_data.get("planId")
self.bucket_id = cloud_data.get("bucketId")
self.title = cloud_data.get(self._cc("title"), "")
self.priority = cloud_data.get(self._cc("priority"), "")
self.assignments = cloud_data.get(self._cc("assignments"), "")
self.order_hint = cloud_data.get(self._cc("orderHint"), "")
self.assignee_priority = cloud_data.get(self._cc("assigneePriority"), "")
self.percent_complete = cloud_data.get(self._cc("percentComplete"), "")
self.has_description = cloud_data.get(self._cc("hasDescription"), "")
created = cloud_data.get(self._cc("createdDateTime"), None)
due_date_time = cloud_data.get(self._cc("dueDateTime"), None)
start_date_time = cloud_data.get(self._cc("startDateTime"), None)
completed_date = cloud_data.get(self._cc("completedDateTime"), None)
local_tz = self.protocol.timezone
self.start_date_time = (
parse(start_date_time).astimezone(local_tz) if start_date_time else None
)
self.created_date = parse(created).astimezone(local_tz) if created else None
self.due_date_time = (
parse(due_date_time).astimezone(local_tz) if due_date_time else None
)
self.completed_date = (
parse(completed_date).astimezone(local_tz) if completed_date else None
)
self.preview_type = cloud_data.get(self._cc("previewType"), None)
self.reference_count = cloud_data.get(self._cc("referenceCount"), None)
self.checklist_item_count = cloud_data.get(self._cc("checklistItemCount"), None)
self.active_checklist_item_count = cloud_data.get(
self._cc("activeChecklistItemCount"), None
)
self.conversation_thread_id = cloud_data.get(
self._cc("conversationThreadId"), None
)
self.applied_categories = cloud_data.get(self._cc("appliedCategories"), None)
self._etag = cloud_data.get("@odata.etag", "")
def __str__(self):
return self.__repr__()
def __repr__(self):
return "Task: {}".format(self.title)
def __eq__(self, other):
return self.object_id == other.object_id
[docs]
def get_details(self):
"""Returns Microsoft O365/AD plan with given id
:rtype: PlanDetails
"""
if not self.object_id:
raise RuntimeError("Plan is not initialized correctly. Id is missing...")
url = self.build_url(
self._endpoints.get("get_details").format(id=self.object_id)
)
response = self.con.get(url)
if not response:
return None
data = response.json()
return self.task_details_constructor(
parent=self,
**{self._cloud_data_key: data},
)
[docs]
def update(self, **kwargs):
"""Updates this task
:param kwargs: all the properties to be updated.
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
url = self.build_url(self._endpoints.get("task").format(id=self.object_id))
for k, v in kwargs.items():
if k in ("start_date_time", "due_date_time"):
kwargs[k] = (
v.strftime("%Y-%m-%dT%H:%M:%SZ")
if isinstance(v, (datetime, date))
else v
)
data = {
self._cc(key): value
for key, value in kwargs.items()
if key
in (
"title",
"priority",
"assignments",
"order_hint",
"assignee_priority",
"percent_complete",
"has_description",
"start_date_time",
"created_date",
"due_date_time",
"completed_date",
"preview_type",
"reference_count",
"checklist_item_count",
"active_checklist_item_count",
"conversation_thread_id",
"applied_categories",
"bucket_id",
)
}
if not data:
return False
response = self.con.patch(
url,
data=data,
headers={"If-Match": self._etag, "Prefer": "return=representation"},
)
if not response:
return False
new_data = response.json()
for key in data:
value = new_data.get(key, None)
if value is not None:
setattr(self, self.protocol.to_api_case(key), value)
self._etag = new_data.get("@odata.etag")
return True
[docs]
def delete(self):
"""Deletes this task
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
url = self.build_url(self._endpoints.get("task").format(id=self.object_id))
response = self.con.delete(url, headers={"If-Match": self._etag})
if not response:
return False
self.object_id = None
return True
[docs]
class Bucket(ApiComponent):
_endpoints = {
"list_tasks": "/planner/buckets/{id}/tasks",
"create_task": "/planner/tasks",
"bucket": "/planner/buckets/{id}",
}
task_constructor = Task
[docs]
def __init__(self, *, parent=None, con=None, **kwargs):
"""A Microsoft O365 bucket
:param parent: parent object
:type parent: Planner or Plan
:param Connection con: connection to use if no parent specified
:param Protocol protocol: protocol to use if no parent specified
(kwargs)
:param str main_resource: use this resource instead of parent resource
(kwargs)
"""
if parent and con:
raise ValueError("Need a parent or a connection but not both")
self.con = parent.con if parent else con
cloud_data = kwargs.get(self._cloud_data_key, {})
self.object_id = cloud_data.get("id")
# Choose the main_resource passed in kwargs over parent main_resource
main_resource = kwargs.pop("main_resource", None) or (
getattr(parent, "main_resource", None) if parent else None
)
main_resource = "{}{}".format(main_resource, "")
super().__init__(
protocol=parent.protocol if parent else kwargs.get("protocol"),
main_resource=main_resource,
)
self.name = cloud_data.get(self._cc("name"), "")
self.order_hint = cloud_data.get(self._cc("orderHint"), "")
self.plan_id = cloud_data.get(self._cc("planId"), "")
self._etag = cloud_data.get("@odata.etag", "")
def __str__(self):
return self.__repr__()
def __repr__(self):
return "Bucket: {}".format(self.name)
def __eq__(self, other):
return self.object_id == other.object_id
[docs]
def list_tasks(self):
"""Returns list of tasks that given plan has
:rtype: list[Task]
"""
if not self.object_id:
raise RuntimeError("Bucket is not initialized correctly. Id is missing...")
url = self.build_url(
self._endpoints.get("list_tasks").format(id=self.object_id)
)
response = self.con.get(url)
if not response:
return None
data = response.json()
return [
self.task_constructor(parent=self, **{self._cloud_data_key: task})
for task in data.get("value", [])
]
[docs]
def create_task(self, title, assignments=None, **kwargs):
"""Creates a Task
:param str title: the title of the task
:param dict assignments: the dict of users to which tasks are to be assigned.
.. code-block:: python
e.g. assignments = {
"ca2a1df2-e36b-4987-9f6b-0ea462f4eb47": null,
"4e98f8f1-bb03-4015-b8e0-19bb370949d8": {
"@odata.type": "microsoft.graph.plannerAssignment",
"orderHint": "String"
}
}
if "user_id": null -> task is unassigned to user.
if "user_id": dict -> task is assigned to user
:param dict kwargs: optional extra parameters to include in the task
:param int priority: priority of the task. The valid range of values is between 0 and 10.
1 -> "urgent", 3 -> "important", 5 -> "medium", 9 -> "low" (kwargs)
:param str order_hint: the order of the bucket. Default is on top (kwargs)
:param datetime or str start_date_time: the starting date of the task. If str format should be: "%Y-%m-%dT%H:%M:%SZ" (kwargs)
:param datetime or str due_date_time: the due date of the task. If str format should be: "%Y-%m-%dT%H:%M:%SZ" (kwargs)
:param str conversation_thread_id: thread ID of the conversation on the task.
This is the ID of the conversation thread object created in the group (kwargs)
:param str assignee_priority: hint used to order items of this type in a list view (kwargs)
:param int percent_complete: percentage of task completion. When set to 100, the task is considered completed (kwargs)
:param dict applied_categories: The categories (labels) to which the task has been applied.
Format should be e.g. {"category1": true, "category3": true, "category5": true } should (kwargs)
:return: newly created task
:rtype: Task
"""
if not title:
raise RuntimeError('Provide a title for the Task')
if not self.object_id and not self.plan_id:
return None
url = self.build_url(
self._endpoints.get('create_task'))
if not assignments:
assignments = {'@odata.type': 'microsoft.graph.plannerAssignments'}
for k, v in kwargs.items():
if k in ('start_date_time', 'due_date_time'):
kwargs[k] = v.strftime('%Y-%m-%dT%H:%M:%SZ') if isinstance(v, (datetime, date)) else v
kwargs = {self._cc(key): value for key, value in kwargs.items() if
key in (
'priority'
'order_hint'
'assignee_priority'
'percent_complete'
'has_description'
'start_date_time'
'created_date'
'due_date_time'
'completed_date'
'preview_type'
'reference_count'
'checklist_item_count'
'active_checklist_item_count'
'conversation_thread_id'
'applied_categories'
)}
data = {
'title': title,
'assignments': assignments,
'bucketId': self.object_id,
'planId': self.plan_id,
**kwargs
}
response = self.con.post(url, data=data)
if not response:
return None
task = response.json()
return self.task_constructor(parent=self,
**{self._cloud_data_key: task})
[docs]
def update(self, **kwargs):
""" Updates this bucket
:param kwargs: all the properties to be updated.
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
url = self.build_url(
self._endpoints.get('bucket').format(id=self.object_id))
data = {self._cc(key): value for key, value in kwargs.items() if
key in ('name', 'order_hint')}
if not data:
return False
response = self.con.patch(url, data=data, headers={'If-Match': self._etag, 'Prefer': 'return=representation'})
if not response:
return False
new_data = response.json()
for key in data:
value = new_data.get(key, None)
if value is not None:
setattr(self, self.protocol.to_api_case(key), value)
self._etag = new_data.get('@odata.etag')
return True
[docs]
def delete(self):
""" Deletes this bucket
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
url = self.build_url(
self._endpoints.get('bucket').format(id=self.object_id))
response = self.con.delete(url, headers={'If-Match': self._etag})
if not response:
return False
self.object_id = None
return True
[docs]
class Plan(ApiComponent):
_endpoints = {
'list_buckets': '/planner/plans/{id}/buckets',
'list_tasks': '/planner/plans/{id}/tasks',
'get_details': '/planner/plans/{id}/details',
'plan': '/planner/plans/{id}',
'create_bucket': '/planner/buckets'
}
bucket_constructor = Bucket
task_constructor = Task
plan_details_constructor = PlanDetails
[docs]
def __init__(self, *, parent=None, con=None, **kwargs):
""" A Microsoft O365 plan
:param parent: parent object
:type parent: Planner
:param Connection con: connection to use if no parent specified
:param Protocol protocol: protocol to use if no parent specified
(kwargs)
:param str main_resource: use this resource instead of parent resource
(kwargs)
"""
if parent and con:
raise ValueError('Need a parent or a connection but not both')
self.con = parent.con if parent else con
cloud_data = kwargs.get(self._cloud_data_key, {})
self.object_id = cloud_data.get('id')
# Choose the main_resource passed in kwargs over parent main_resource
main_resource = kwargs.pop('main_resource', None) or (
getattr(parent, 'main_resource', None) if parent else None)
main_resource = '{}{}'.format(main_resource, '')
super().__init__(
protocol=parent.protocol if parent else kwargs.get('protocol'),
main_resource=main_resource)
self.created_date_time = cloud_data.get(self._cc('createdDateTime'), '')
container = cloud_data.get(self._cc('container'), {})
self.group_id = container.get(self._cc('containerId'), '')
self.title = cloud_data.get(self._cc('title'), '')
self._etag = cloud_data.get('@odata.etag', '')
def __str__(self):
return self.__repr__()
def __repr__(self):
return 'Plan: {}'.format(self.title)
def __eq__(self, other):
return self.object_id == other.object_id
[docs]
def list_buckets(self):
""" Returns list of buckets that given plan has
:rtype: list[Bucket]
"""
if not self.object_id:
raise RuntimeError('Plan is not initialized correctly. Id is missing...')
url = self.build_url(
self._endpoints.get('list_buckets').format(id=self.object_id))
response = self.con.get(url)
if not response:
return None
data = response.json()
return [
self.bucket_constructor(parent=self, **{self._cloud_data_key: bucket})
for bucket in data.get('value', [])]
[docs]
def list_tasks(self):
""" Returns list of tasks that given plan has
:rtype: list[Task] or Pagination of Task
"""
if not self.object_id:
raise RuntimeError('Plan is not initialized correctly. Id is missing...')
url = self.build_url(
self._endpoints.get('list_tasks').format(id=self.object_id))
response = self.con.get(url)
if not response:
return []
data = response.json()
next_link = data.get(NEXT_LINK_KEYWORD, None)
tasks = [
self.task_constructor(parent=self, **{self._cloud_data_key: task})
for task in data.get('value', [])]
if next_link:
return Pagination(parent=self, data=tasks,
constructor=self.task_constructor,
next_link=next_link)
else:
return tasks
[docs]
def get_details(self):
""" Returns Microsoft O365/AD plan with given id
:rtype: PlanDetails
"""
if not self.object_id:
raise RuntimeError('Plan is not initialized correctly. Id is missing...')
url = self.build_url(
self._endpoints.get('get_details').format(id=self.object_id))
response = self.con.get(url)
if not response:
return None
data = response.json()
return self.plan_details_constructor(parent=self,
**{self._cloud_data_key: data}, )
[docs]
def create_bucket(self, name, order_hint=' !'):
""" Creates a Bucket
:param str name: the name of the bucket
:param str order_hint: the order of the bucket. Default is on top.
How to use order hints here: https://docs.microsoft.com/en-us/graph/api/resources/planner-order-hint-format?view=graph-rest-1.0
:return: newly created bucket
:rtype: Bucket
"""
if not name:
raise RuntimeError('Provide a name for the Bucket')
if not self.object_id:
return None
url = self.build_url(
self._endpoints.get('create_bucket'))
data = {'name': name, 'orderHint': order_hint, 'planId': self.object_id}
response = self.con.post(url, data=data)
if not response:
return None
bucket = response.json()
return self.bucket_constructor(parent=self,
**{self._cloud_data_key: bucket})
[docs]
def update(self, **kwargs):
""" Updates this plan
:param kwargs: all the properties to be updated.
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
url = self.build_url(
self._endpoints.get('plan').format(id=self.object_id))
data = {self._cc(key): value for key, value in kwargs.items() if
key in ('title')}
if not data:
return False
response = self.con.patch(url, data=data, headers={'If-Match': self._etag, 'Prefer': 'return=representation'})
if not response:
return False
new_data = response.json()
for key in data:
value = new_data.get(key, None)
if value is not None:
setattr(self, self.protocol.to_api_case(key), value)
self._etag = new_data.get('@odata.etag')
return True
[docs]
def delete(self):
""" Deletes this plan
:return: Success / Failure
:rtype: bool
"""
if not self.object_id:
return False
url = self.build_url(
self._endpoints.get('plan').format(id=self.object_id))
response = self.con.delete(url, headers={'If-Match': self._etag})
if not response:
return False
self.object_id = None
return True
[docs]
class Planner(ApiComponent):
""" A microsoft planner class
In order to use the API following permissions are required.
Delegated (work or school account) - Group.Read.All, Group.ReadWrite.All
"""
_endpoints = {
'get_my_tasks': '/me/planner/tasks',
'get_plan_by_id': '/planner/plans/{plan_id}',
'get_bucket_by_id': '/planner/buckets/{bucket_id}',
'get_task_by_id': '/planner/tasks/{task_id}',
'list_user_tasks': '/users/{user_id}/planner/tasks',
'list_group_plans': '/groups/{group_id}/planner/plans',
'create_plan': '/planner/plans',
}
plan_constructor = Plan
bucket_constructor = Bucket
task_constructor = Task
[docs]
def __init__(self, *, parent=None, con=None, **kwargs):
""" A Planner object
:param parent: parent object
:type parent: Account
:param Connection con: connection to use if no parent specified
:param Protocol protocol: protocol to use if no parent specified
(kwargs)
:param str main_resource: use this resource instead of parent resource
(kwargs)
"""
if parent and con:
raise ValueError('Need a parent or a connection but not both')
self.con = parent.con if parent else con
# Choose the main_resource passed in kwargs over the host_name
main_resource = kwargs.pop('main_resource',
'') # defaults to blank resource
super().__init__(
protocol=parent.protocol if parent else kwargs.get('protocol'),
main_resource=main_resource)
def __str__(self):
return self.__repr__()
def __repr__(self):
return 'Microsoft Planner'
[docs]
def get_my_tasks(self, *args):
""" Returns a list of open planner tasks assigned to me
:rtype: tasks
"""
url = self.build_url(self._endpoints.get('get_my_tasks'))
response = self.con.get(url)
if not response:
return None
data = response.json()
return [
self.task_constructor(parent=self, **{self._cloud_data_key: site})
for site in data.get('value', [])]
[docs]
def get_plan_by_id(self, plan_id=None):
""" Returns Microsoft O365/AD plan with given id
:param plan_id: plan id of plan
:rtype: Plan
"""
if not plan_id:
raise RuntimeError('Provide the plan_id')
url = self.build_url(
self._endpoints.get('get_plan_by_id').format(plan_id=plan_id))
response = self.con.get(url)
if not response:
return None
data = response.json()
return self.plan_constructor(parent=self,
**{self._cloud_data_key: data}, )
[docs]
def get_bucket_by_id(self, bucket_id=None):
""" Returns Microsoft O365/AD plan with given id
:param bucket_id: bucket id of buckets
:rtype: Bucket
"""
if not bucket_id:
raise RuntimeError('Provide the bucket_id')
url = self.build_url(
self._endpoints.get('get_bucket_by_id').format(bucket_id=bucket_id))
response = self.con.get(url)
if not response:
return None
data = response.json()
return self.bucket_constructor(parent=self,
**{self._cloud_data_key: data})
[docs]
def get_task_by_id(self, task_id=None):
""" Returns Microsoft O365/AD plan with given id
:param task_id: task id of tasks
:rtype: Task
"""
if not task_id:
raise RuntimeError('Provide the task_id')
url = self.build_url(
self._endpoints.get('get_task_by_id').format(task_id=task_id))
response = self.con.get(url)
if not response:
return None
data = response.json()
return self.task_constructor(parent=self,
**{self._cloud_data_key: data})
[docs]
def list_user_tasks(self, user_id=None):
""" Returns Microsoft O365/AD plan with given id
:param user_id: user id
:rtype: list[Task]
"""
if not user_id:
raise RuntimeError('Provide the user_id')
url = self.build_url(
self._endpoints.get('list_user_tasks').format(user_id=user_id))
response = self.con.get(url)
if not response:
return None
data = response.json()
return [
self.task_constructor(parent=self, **{self._cloud_data_key: task})
for task in data.get('value', [])]
[docs]
def list_group_plans(self, group_id=None):
""" Returns list of plans that given group has
:param group_id: group id
:rtype: list[Plan]
"""
if not group_id:
raise RuntimeError('Provide the group_id')
url = self.build_url(
self._endpoints.get('list_group_plans').format(group_id=group_id))
response = self.con.get(url)
if not response:
return None
data = response.json()
return [
self.plan_constructor(parent=self, **{self._cloud_data_key: plan})
for plan in data.get('value', [])]
[docs]
def create_plan(self, owner, title='Tasks'):
""" Creates a Plan
:param str owner: the id of the group that will own the plan
:param str title: the title of the new plan. Default set to "Tasks"
:return: newly created plan
:rtype: Plan
"""
if not owner:
raise RuntimeError('Provide the owner (group_id)')
url = self.build_url(
self._endpoints.get('create_plan'))
data = {'owner': owner, 'title': title}
response = self.con.post(url, data=data)
if not response:
return None
plan = response.json()
return self.plan_constructor(parent=self,
**{self._cloud_data_key: plan})