MOOOARRR TYPEZZZZZZ HINTZZZZ

Author: Sebastian Molenda

pubnub/endpoints/access/grant_token.py

@@ -1,58 +1,77 @@
+from typing import Union, List, Optional
 from pubnub import utils
 from pubnub.endpoints.endpoint import Endpoint
 from pubnub.errors import PNERR_TTL_MISSING, PNERR_INVALID_META, PNERR_RESOURCES_MISSING
 from pubnub.exceptions import PubNubException
 from pubnub.enums import HttpMethod, PNOperationType
+from pubnub.models.consumer.common import PNStatus
 from pubnub.models.consumer.v3.access_manager import PNGrantTokenResult
+from pubnub.structures import Envelope
+
+
+class PNGrantTokenResultEnvelope(Envelope):
+    result: PNGrantTokenResult
+    status: PNStatus
 
 
 class GrantToken(Endpoint):
     GRANT_TOKEN_PATH = "/v3/pam/%s/grant"
 
-    def __init__(self, pubnub):
+    def __init__(self, pubnub, channels: Union[str, List[str]] = None, channel_groups: Union[str, List[str]] = None,
+                 users: Union[str, List[str]] = None, spaces: Union[str, List[str]] = None,
+                 authorized_user_id: str = None, ttl: Optional[int] = None, meta: Optional[any] = None):
         Endpoint.__init__(self, pubnub)
-        self._ttl = None
-        self._meta = None
-        self._authorized_uuid = None
+        self._ttl = ttl
+        self._meta = meta
+        self._authorized_uuid = authorized_user_id
         self._channels = []
+        if channels:
+            utils.extend_list(self._channels, channels)
+        if spaces:
+            utils.extend_list(self._channels, spaces)
+
         self._groups = []
+        if channel_groups:
+            utils.extend_list(self._groups, channel_groups)
         self._uuids = []
+        if users:
+            utils.extend_list(self._uuids, users)
 
         self._sort_params = True
 
-    def ttl(self, ttl):
+    def ttl(self, ttl: int) -> 'GrantToken':
         self._ttl = ttl
         return self
 
-    def meta(self, meta):
+    def meta(self, meta: any) -> 'GrantToken':
         self._meta = meta
         return self
 
-    def authorized_uuid(self, uuid):
+    def authorized_uuid(self, uuid: str) -> 'GrantToken':
         self._authorized_uuid = uuid
         return self
 
-    def authorized_user(self, user):
+    def authorized_user(self, user) -> 'GrantToken':
         self._authorized_uuid = user
         return self
 
-    def spaces(self, spaces):
+    def spaces(self, spaces: Union[str, List[str]]) -> 'GrantToken':
         self._channels = spaces
         return self
 
-    def users(self, users):
+    def users(self, users: Union[str, List[str]]) -> 'GrantToken':
         self._uuids = users
         return self
 
-    def channels(self, channels):
+    def channels(self, channels: Union[str, List[str]]) -> 'GrantToken':
         self._channels = channels
         return self
 
-    def groups(self, groups):
+    def groups(self, groups: Union[str, List[str]]) -> 'GrantToken':
         self._groups = groups
         return self
 
-    def uuids(self, uuids):
+    def uuids(self, uuids: Union[str, List[str]]) -> 'GrantToken':
         self._uuids = uuids
         return self
 
@@ -102,9 +121,12 @@ def validate_params(self):
         self.validate_ttl()
         self.validate_resources()
 
-    def create_response(self, envelope):
+    def create_response(self, envelope) -> PNGrantTokenResult:
         return PNGrantTokenResult.from_json(envelope['data'])
 
+    def sync(self):
+        return super().sync()
+
     def is_auth_required(self):
         return False
 

pubnub/endpoints/access/revoke_token.py

@@ -1,13 +1,20 @@
 from pubnub.enums import PNOperationType, HttpMethod
 from pubnub.endpoints.endpoint import Endpoint
+from pubnub.models.consumer.common import PNStatus
 from pubnub.models.consumer.v3.access_manager import PNRevokeTokenResult
 from pubnub import utils
+from pubnub.structures import Envelope
+
+
+class PNRevokeTokenResultEnvelope(Envelope):
+    result: PNRevokeTokenResult
+    status: PNStatus
 
 
 class RevokeToken(Endpoint):
     REVOKE_TOKEN_PATH = "/v3/pam/%s/grant/%s"
 
-    def __init__(self, pubnub, token):
+    def __init__(self, pubnub, token: str):
         Endpoint.__init__(self, pubnub)
         self.token = token
 
@@ -18,6 +25,9 @@ def validate_params(self):
     def create_response(self, envelope):
         return PNRevokeTokenResult(envelope)
 
+    def sync(self) -> PNRevokeTokenResultEnvelope:
+        return PNRevokeTokenResultEnvelope(super().sync())
+
     def is_auth_required(self):
         return False
 

pubnub/endpoints/channel_groups/add_channel_to_channel_group.py

@@ -1,31 +1,36 @@
+from typing import List, Union
 from pubnub import utils
 from pubnub.endpoints.endpoint import Endpoint
 from pubnub.errors import PNERR_CHANNELS_MISSING, PNERR_GROUP_MISSING
 from pubnub.exceptions import PubNubException
 from pubnub.enums import HttpMethod, PNOperationType
 from pubnub.models.consumer.channel_group import PNChannelGroupsAddChannelResult
+from pubnub.models.consumer.common import PNStatus
+from pubnub.structures import Envelope
+
+
+class PNChannelGroupsAddChannelResultEnvelope(Envelope):
+    result: PNChannelGroupsAddChannelResult
+    status: PNStatus
 
 
 class AddChannelToChannelGroup(Endpoint):
     # /v1/channel-registration/sub-key/<sub_key>/channel-group/<group_name>?add=ch1,ch2
     ADD_PATH = "/v1/channel-registration/sub-key/%s/channel-group/%s"
 
-    def __init__(self, pubnub):
+    def __init__(self, pubnub, channels: Union[str, List[str]] = None, channel_group: str = None):
         Endpoint.__init__(self, pubnub)
         self._channels = []
-        self._channel_group = None
-
-    def channels(self, channels):
-        if isinstance(channels, (list, tuple)):
-            self._channels.extend(channels)
-        else:
-            self._channels.extend(utils.split_items(channels))
+        if channels:
+            utils.extend_list(self._channels, channels)
+        self._channel_group = channel_group
 
+    def channels(self, channels) -> 'AddChannelToChannelGroup':
+        utils.extend_list(self._channels, channels)
         return self
 
-    def channel_group(self, channel_group):
+    def channel_group(self, channel_group: str) -> 'AddChannelToChannelGroup':
         self._channel_group = channel_group
-
         return self
 
     def custom_params(self):
@@ -50,9 +55,12 @@ def validate_params(self):
     def is_auth_required(self):
         return True
 
-    def create_response(self, envelope):
+    def create_response(self, envelope) -> PNChannelGroupsAddChannelResult:
         return PNChannelGroupsAddChannelResult()
 
+    def sync(self):
+        return PNChannelGroupsAddChannelResultEnvelope(super().sync())
+
     def request_timeout(self):
         return self.pubnub.config.non_subscribe_request_timeout
 

pubnub/endpoints/channel_groups/list_channels_in_channel_group.py

@@ -4,19 +4,25 @@
 from pubnub.exceptions import PubNubException
 from pubnub.enums import HttpMethod, PNOperationType
 from pubnub.models.consumer.channel_group import PNChannelGroupsListResult
+from pubnub.models.consumer.common import PNStatus
+from pubnub.structures import Envelope
+
+
+class PNChannelGroupsListResultEnvelope(Envelope):
+    result: PNChannelGroupsListResult
+    status: PNStatus
 
 
 class ListChannelsInChannelGroup(Endpoint):
     # /v1/channel-registration/sub-key/<sub_key>/channel-group/<group_name>
     LIST_PATH = "/v1/channel-registration/sub-key/%s/channel-group/%s"
 
-    def __init__(self, pubnub):
+    def __init__(self, pubnub, channel_group: str = None):
         Endpoint.__init__(self, pubnub)
-        self._channel_group = None
-
-    def channel_group(self, channel_group):
         self._channel_group = channel_group
 
+    def channel_group(self, channel_group: str) -> 'ListChannelsInChannelGroup':
+        self._channel_group = channel_group
         return self
 
     def custom_params(self):
@@ -35,12 +41,15 @@ def validate_params(self):
         if not isinstance(self._channel_group, str) or len(self._channel_group) == 0:
             raise PubNubException(pn_error=PNERR_GROUP_MISSING)
 
-    def create_response(self, envelope):
+    def create_response(self, envelope) -> PNChannelGroupsListResult:
         if 'payload' in envelope and 'channels' in envelope['payload']:
             return PNChannelGroupsListResult(envelope['payload']['channels'])
         else:
             return PNChannelGroupsListResult([])
 
+    def sync(self) -> PNChannelGroupsListResultEnvelope:
+        return PNChannelGroupsListResultEnvelope(super().sync())
+
     def is_auth_required(self):
         return True
 

pubnub/endpoints/channel_groups/remove_channel_from_channel_group.py

@@ -1,31 +1,38 @@
+from typing import List, Union
 from pubnub import utils
 from pubnub.endpoints.endpoint import Endpoint
 from pubnub.errors import PNERR_CHANNELS_MISSING, PNERR_GROUP_MISSING
 from pubnub.exceptions import PubNubException
 from pubnub.enums import HttpMethod, PNOperationType
 from pubnub.models.consumer.channel_group import PNChannelGroupsRemoveChannelResult
+from pubnub.models.consumer.common import PNStatus
+from pubnub.structures import Envelope
+
+
+class PNChannelGroupsRemoveChannelResultEnvelope(Envelope):
+    result: PNChannelGroupsRemoveChannelResult
+    status: PNStatus
 
 
 class RemoveChannelFromChannelGroup(Endpoint):
     # /v1/channel-registration/sub-key/<sub_key>/channel-group/<group_name>?remove=ch1,ch2
     REMOVE_PATH = "/v1/channel-registration/sub-key/%s/channel-group/%s"
+    _channels: list = []
+    _channel_group: str = None
 
-    def __init__(self, pubnub):
+    def __init__(self, pubnub, channels: Union[str, List[str]] = None, channel_group: str = None):
         Endpoint.__init__(self, pubnub)
         self._channels = []
-        self._channel_group = None
-
-    def channels(self, channels):
-        if isinstance(channels, (list, tuple)):
-            self._channels.extend(channels)
-        else:
-            self._channels.extend(utils.split_items(channels))
+        if channels:
+            utils.extend_list(self._channels, channels)
+        self._channel_group = channel_group
 
+    def channels(self, channels) -> 'RemoveChannelFromChannelGroup':
+        utils.extend_list(self._channels, channels)
         return self
 
-    def channel_group(self, channel_group):
+    def channel_group(self, channel_group: str) -> 'RemoveChannelFromChannelGroup':
         self._channel_group = channel_group
-
         return self
 
     def custom_params(self):
@@ -50,9 +57,12 @@ def validate_params(self):
     def is_auth_required(self):
         return True
 
-    def create_response(self, envelope):
+    def create_response(self, envelope) -> PNChannelGroupsRemoveChannelResult:
         return PNChannelGroupsRemoveChannelResult()
 
+    def sync(self):
+        return PNChannelGroupsRemoveChannelResultEnvelope(super().sync())
+
     def request_timeout(self):
         return self.pubnub.config.non_subscribe_request_timeout
 

pubnub/endpoints/channel_groups/remove_channel_group.py

@@ -4,19 +4,25 @@
 from pubnub.exceptions import PubNubException
 from pubnub.enums import HttpMethod, PNOperationType
 from pubnub.models.consumer.channel_group import PNChannelGroupsRemoveGroupResult
+from pubnub.models.consumer.common import PNStatus
+from pubnub.structures import Envelope
+
+
+class PNChannelGroupsRemoveGroupResultEnvelope(Envelope):
+    result: PNChannelGroupsRemoveGroupResult
+    status: PNStatus
 
 
 class RemoveChannelGroup(Endpoint):
     # /v1/channel-registration/sub-key/<sub_key>/channel-group/<group_name>/remove
     REMOVE_PATH = "/v1/channel-registration/sub-key/%s/channel-group/%s/remove"
 
-    def __init__(self, pubnub):
+    def __init__(self, pubnub, channel_group: str = None):
         Endpoint.__init__(self, pubnub)
-        self._channel_group = None
-
-    def channel_group(self, channel_group):
         self._channel_group = channel_group
 
+    def channel_group(self, channel_group: str) -> 'RemoveChannelGroup':
+        self._channel_group = channel_group
         return self
 
     def custom_params(self):
@@ -41,6 +47,9 @@ def is_auth_required(self):
     def create_response(self, envelope):
         return PNChannelGroupsRemoveGroupResult()
 
+    def sync(self) -> PNChannelGroupsRemoveGroupResultEnvelope:
+        return PNChannelGroupsRemoveGroupResultEnvelope(super().sync())
+
     def request_timeout(self):
         return self.pubnub.config.non_subscribe_request_timeout