ci: regenerated with OpenAPI Doc 0.0.3, Speakeasy CLI 1.161.0

src/plex_api/activities.py

@@ -0,0 +1,104 @@
+"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
+
+from .sdkconfiguration import SDKConfiguration
+from plex_api import utils
+from plex_api.models import errors, operations
+from typing import Optional
+
+class Activities:
+    r"""Activities are awesome. They provide a way to monitor and control asynchronous operations on the server. In order to receive real-time updates for activities, a client would normally subscribe via either EventSource or Websocket endpoints.
+    Activities are associated with HTTP replies via a special `X-Plex-Activity` header which contains the UUID of the activity.
+    Activities are optional cancellable. If cancellable, they may be cancelled via the `DELETE` endpoint. Other details:
+    - They can contain a `progress` (from 0 to 100) marking the percent completion of the activity.
+    - They must contain an `type` which is used by clients to distinguish the specific activity.
+    - They may contain a `Context` object with attributes which associate the activity with various specific entities (items, libraries, etc.)
+    - The may contain a `Response` object which attributes which represent the result of the asynchronous operation.
+    """
+    sdk_configuration: SDKConfiguration
+
+    def __init__(self, sdk_config: SDKConfiguration) -> None:
+        self.sdk_configuration = sdk_config
+        
+    
+    
+    def get_server_activities(self) -> operations.GetServerActivitiesResponse:
+        r"""Get Server Activities
+        Get Server Activities
+        """
+        base_url = utils.template_url(*self.sdk_configuration.get_server_details())
+        
+        url = base_url + '/activities'
+        headers = {}
+        headers['Accept'] = 'application/json'
+        headers['user-agent'] = self.sdk_configuration.user_agent
+        
+        if callable(self.sdk_configuration.security):
+            client = utils.configure_security_client(self.sdk_configuration.client, self.sdk_configuration.security())
+        else:
+            client = utils.configure_security_client(self.sdk_configuration.client, self.sdk_configuration.security)
+        
+        http_res = client.request('GET', url, headers=headers)
+        content_type = http_res.headers.get('Content-Type')
+        
+        res = operations.GetServerActivitiesResponse(status_code=http_res.status_code, content_type=content_type, raw_response=http_res)
+        
+        if http_res.status_code == 200:
+            if utils.match_content_type(content_type, 'application/json'):
+                out = utils.unmarshal_json(http_res.text, Optional[operations.GetServerActivitiesResponseBody])
+                res.object = out
+            else:
+                raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
+        elif http_res.status_code == 400 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
+            raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
+        elif http_res.status_code == 401:
+            if utils.match_content_type(content_type, 'application/json'):
+                out = utils.unmarshal_json(http_res.text, errors.GetServerActivitiesResponseBody)
+                out.raw_response = http_res
+                raise out
+            else:
+                raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
+
+        return res
+
+    
+    
+    def cancel_server_activities(self, activity_uuid: str) -> operations.CancelServerActivitiesResponse:
+        r"""Cancel Server Activities
+        Cancel Server Activities
+        """
+        request = operations.CancelServerActivitiesRequest(
+            activity_uuid=activity_uuid,
+        )
+        
+        base_url = utils.template_url(*self.sdk_configuration.get_server_details())
+        
+        url = utils.generate_url(operations.CancelServerActivitiesRequest, base_url, '/activities/{activityUUID}', request)
+        headers = {}
+        headers['Accept'] = 'application/json'
+        headers['user-agent'] = self.sdk_configuration.user_agent
+        
+        if callable(self.sdk_configuration.security):
+            client = utils.configure_security_client(self.sdk_configuration.client, self.sdk_configuration.security())
+        else:
+            client = utils.configure_security_client(self.sdk_configuration.client, self.sdk_configuration.security)
+        
+        http_res = client.request('DELETE', url, headers=headers)
+        content_type = http_res.headers.get('Content-Type')
+        
+        res = operations.CancelServerActivitiesResponse(status_code=http_res.status_code, content_type=content_type, raw_response=http_res)
+        
+        if http_res.status_code == 200:
+            pass
+        elif http_res.status_code == 400 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
+            raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
+        elif http_res.status_code == 401:
+            if utils.match_content_type(content_type, 'application/json'):
+                out = utils.unmarshal_json(http_res.text, errors.CancelServerActivitiesResponseBody)
+                out.raw_response = http_res
+                raise out
+            else:
+                raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
+
+        return res
+
+