Source code for O365.planner

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})