Merge pull request #17 from krcb197/12-add-ability-to-update-and-event-start-and-end

Add the ability to duplicate and delete events

Author: krcb197

src/pytito/__about__.py

@@ -17,4 +17,4 @@
 
 Variables that describes the Package
 """
-__version__ = "0.0.10"
+__version__ = "0.0.11"

src/pytito/admin/_base_client.py

@@ -110,7 +110,7 @@ def _get_response(self, endpoint: str) -> dict[str, Any]:
 
         return response.json()
 
-    def _patch_reponse(self, value: dict[str, Any]) -> None:
+    def _patch_response(self, value: dict[str, Any]) -> None:
 
         response = requests.patch(
             url=self._end_point,
@@ -130,6 +130,47 @@ def _patch_reponse(self, value: dict[str, Any]) -> None:
         if not response.status_code == 200:
             raise RuntimeError(f'patch failed with status code: {response.status_code}')
 
+    def _post_response(self, endpoint: str, value: dict[str, Any]) -> None:
+
+        if endpoint == '':
+            full_end_point = self._end_point
+        else:
+            full_end_point = self._end_point + '/' + endpoint
+
+        response = requests.post(
+            url=full_end_point,
+            headers={"Accept" : "application/json",
+                     "Authorization" : f"Token token={self.__api_key()}"},
+            json=value,
+            timeout=10.0
+        )
+
+        if response.status_code == 401:
+            raise UnauthorizedException(response.json()['message'])
+
+        if response.status_code == 403:
+            detail = json.loads(response.text)
+            raise ForbiddenException(detail['errors']['detail'])
+
+        if response.status_code not in [200, 201]:
+            raise RuntimeError(f'post failed with status code: {response.status_code}')
+
+    def _delete_response(self) -> None:
+
+        response = requests.patch(
+            url=self._end_point,
+            headers={"Accept" : "application/json",
+                     "Authorization" : f"Token token={self.__api_key()}"},
+            timeout=10.0
+        )
+
+        if response.status_code == 401:
+            raise UnauthorizedException(response.json()['message'])
+
+        if response.status_code == 403:
+            detail = json.loads(response.text)
+            raise ForbiddenException(detail['errors']['detail'])
+
 
 
 class EventChildAPIBase(AdminAPIBase, ABC):

src/pytito/admin/account.py

@@ -74,7 +74,13 @@ def next_event(self) -> Event:
         """
         Return the chronologically first of the upcoming events
         """
-        upcoming_events = list(self.events.values())
+
+        # in some case draft event may not have a start date configured so must be excluded
+        def include_event(event: Event) -> bool:
+            # pylint:disable-next=protected-access
+            return event._json_content['start_at'] is not None
+
+        upcoming_events = list(filter(include_event, self.events.values()))
         upcoming_events.sort(key=attrgetter('start_at'))
         return upcoming_events[0]
 

src/pytito/admin/activity.py

@@ -56,7 +56,7 @@ def _populate_json(self) -> None:
             raise ValueError('expected the extended view of the ticket')
 
     def _update(self, payload: dict[str, Any]) -> None:
-        self._patch_reponse(value={'activity': payload})
+        self._patch_response(value={'activity': payload})
         for key, value in payload.items():
             self._json_content[key] = value
 
@@ -91,7 +91,7 @@ def start_at(self, value: Optional[datetime]) -> None:
                                    'set the end_at to None first')
             payload = {'date': None,
                        'start_time': None}
-            self._patch_reponse(value={'activity': payload})
+            self._patch_response(value={'activity': payload})
             self._json_content['start_at'] = None
         else:
             if self.end_at is not None and self.end_at.date() != value.date():
@@ -103,7 +103,7 @@ def start_at(self, value: Optional[datetime]) -> None:
             # date and time
             payload = {'date': value.strftime("%Y-%m-%d"),
                        'start_time': value.strftime("%H:%M")}
-            self._patch_reponse(value={'activity': payload})
+            self._patch_response(value={'activity': payload})
             value_str = datetime_to_json(value)
             self._json_content['start_at'] = value_str
 
@@ -124,7 +124,7 @@ def end_at(self, value: Optional[datetime]) -> None:
         payload: dict[str, Any]
         if value is None:
             payload = {'end_time': None}
-            self._patch_reponse(value={'activity': payload})
+            self._patch_response(value={'activity': payload})
             self._json_content['end_at'] = None
         else:
             if self.start_at is None:
@@ -137,6 +137,6 @@ def end_at(self, value: Optional[datetime]) -> None:
             # the start_at can not be changed directly, instead it is necessary to modify the
             # date and time
             payload = {'end_time': value.strftime("%H:%M")}
-            self._patch_reponse(value={'activity': payload})
+            self._patch_response(value={'activity': payload})
             value_str = datetime_to_json(value)
             self._json_content['end_at'] = value_str

src/pytito/admin/event.py

@@ -18,6 +18,7 @@
 This file provides the event class
 """
 from typing import Optional, Any
+import time
 
 from datetime import datetime
 
@@ -32,12 +33,15 @@ class Event(AdminAPIBase):
     One of the events available through the Tito IO AdminAPI
     """
 
-    def __init__(self, account_slug:str, event_slug:str,
+    def __init__(self, *, account_slug:str, event_slug:str,
                  json_content:Optional[dict[str, Any]]=None,
-                 api_key: Optional[str] = None) -> None:
-        super().__init__(json_content=json_content, api_key=api_key)
+                 api_key: Optional[str] = None,
+                 allow_automatic_json_retrieval:bool=False) -> None:
+        super().__init__(json_content=json_content, api_key=api_key,
+                         allow_automatic_json_retrieval=allow_automatic_json_retrieval)
         self.__account_slug = account_slug
         self.__event_slug = event_slug
+        self.__api_key_internal = api_key
         if json_content is not None:
             if self._json_content['_type'] != "event":
                 raise ValueError('JSON content type was expected to be ticket')
@@ -60,17 +64,35 @@ def _populate_json(self) -> None:
             raise ValueError('JSON content type was expected to be ticket')
 
     def _update(self, payload: dict[str, Any]) -> None:
-        self._patch_reponse(value={'event': payload})
+        self._patch_response(value={'event': payload})
         for key, value in payload.items():
             self._json_content[key] = value
 
+    def _update_slug(self, new_slug: str) -> None:
+        """
+        The Slug is a unique component of the data used to reference the release in the API.
+        It is sometimes desirable to change this
+
+        .. Warning::
+            Changing the slug may break things, especially if it clashes with another slug.
+            Use this method with caution. In particular, the slug is used to key other
+            dictionaries within the data model. Once changing the clug it is recommended that
+            the whole data model is refreshed
+        """
+        self._update({'slug': new_slug})
+        self.__event_slug = new_slug
+
     @property
     def title(self) -> str:
         """
         Event title
         """
         return self._json_content['title']
 
+    @title.setter
+    def title(self, value: str) -> None:
+        self._update({'title': value})
+
     def __ticket_getter(self) -> list[Ticket]:
 
         def ticket_factory(json_content:dict[str, Any]) -> Ticket:
@@ -104,7 +126,7 @@ def start_at(self, value: datetime) -> None:
         # date and time
         payload = {'start_date': value.strftime("%Y-%m-%d"),
                    'start_time': value.strftime("%H:%M")}
-        self._patch_reponse(value={'event': payload})
+        self._patch_response(value={'event': payload})
         value_str = datetime_to_json(value)
         self._json_content['start_at'] = value_str
 
@@ -124,7 +146,7 @@ def end_at(self, value: datetime) -> None:
         # date and time
         payload = {'end_date': value.strftime("%Y-%m-%d"),
                    'end_time': value.strftime("%H:%M")}
-        self._patch_reponse(value={'event': payload})
+        self._patch_response(value={'event': payload})
         value_str = datetime_to_json(value)
         self._json_content['end_at'] = value_str
 
@@ -177,3 +199,55 @@ def test_mode(self) -> bool:
         Whether the event is in test mode
         """
         return self._json_content['test_mode']
+
+    def duplicate_event(self, title:str, slug:Optional[str]=None) -> "Event":
+        """
+        Duplicate the event and then update the title and optionally the new slug for the
+        created event
+        :param title: New event title
+        :param slug: New event slug, a value of None will leave the automatically created slug in
+                     place
+        :return: The newly created event
+        """
+        self._post_response('duplication', value={})
+        for _ in range(120):
+            time.sleep(1)
+            duplication_status = self._get_duplication_status()
+            status = duplication_status['status']
+            if status == 'processing':
+                # pylint:disable-next=bad-builtin
+                print('Duplication in progress')
+                continue
+            if status == 'complete':
+                new_slug = duplication_status['slug']
+                new_title = duplication_status['title']
+                new_event = Event(account_slug=self.__account_slug,
+                                  event_slug=new_slug,
+                                  json_content=None,
+                                  api_key=self.__api_key_internal,
+                                  allow_automatic_json_retrieval=True)
+                if new_event.title != new_title:
+                    raise ValueError(f'New event has different title to reported value:{new_title}')
+                new_event.title = title
+                if slug is not None:
+                    # The update slug method is a powerful option that is not normally exposed
+                    # to the users so is private
+                    # pylint:disable-next=protected-access
+                    new_event._update_slug(slug)
+                return new_event
+
+            raise ValueError('Unhandled {status=}')
+
+        raise RuntimeError('Timeout During Event Duplication')
+
+    def _get_duplication_status(self) -> dict[str, Any]:
+        duplication_status = self._get_response('duplication')['duplication']
+        if duplication_status['_type'] != '_duplication':
+            raise RuntimeError('Duplication response does not have a value of _type=_duplication')
+        return duplication_status
+
+    def _delete_event(self) -> None:
+        """
+        Delete the event
+        """
+        self._delete_response()

src/pytito/admin/release.py

@@ -58,7 +58,7 @@ def _populate_json(self) -> None:
             raise ValueError('JSON content type was expected to be release')
 
     def _update(self, payload: dict[str, Any]) -> None:
-        self._patch_reponse(value={'release': payload})
+        self._patch_response(value={'release': payload})
         for key, value in payload.items():
             self._json_content[key] = value