Publish, Fire

Author: Sebastian Molenda

pubnub/endpoints/pubsub/fire.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from pubnub import utils
 from pubnub.endpoints.endpoint import Endpoint
 from pubnub.enums import HttpMethod, PNOperationType
@@ -11,33 +12,39 @@ class Fire(Endpoint):
     FIRE_GET_PATH = "/publish/%s/%s/0/%s/%s/%s"
     FIRE_POST_PATH = "/publish/%s/%s/0/%s/%s"
 
-    def __init__(self, pubnub):
+    _channel: str
+    _message: any
+    _use_post: Optional[bool]
+    _meta: Optional[any]
+
+    def __init__(self, pubnub, channel: Optional[str] = None, message: Optional[any] = None,
+                 use_post: Optional[bool] = None, meta: Optional[any] = None):
         Endpoint.__init__(self, pubnub)
-        self._channel = None
-        self._message = None
-        self._use_post = None
-        self._meta = None
+        self._channel = channel
+        self._message = message
+        self._use_post = use_post
+        self._meta = meta
 
-    def channel(self, channel):
+    def channel(self, channel: str) -> 'Fire':
         self._channel = str(channel)
         return self
 
-    def message(self, message):
+    def message(self, message) -> 'Fire':
         self._message = message
         return self
 
-    def use_post(self, use_post):
+    def use_post(self, use_post) -> 'Fire':
         self._use_post = bool(use_post)
         return self
 
-    def is_compressable(self):
+    def is_compressable(self) -> bool:
         return True
 
-    def use_compression(self, compress=True):
+    def use_compression(self, compress=True) -> 'Fire':
         self._use_compression = bool(compress)
         return self
 
-    def meta(self, meta):
+    def meta(self, meta) -> 'Fire':
         self._meta = meta
         return self
 

pubnub/endpoints/pubsub/publish.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from pubnub import utils
 from pubnub.endpoints.endpoint import Endpoint
 from pubnub.errors import PNERR_MESSAGE_MISSING
@@ -12,45 +13,56 @@ class Publish(Endpoint, TimeTokenOverrideMixin):
     PUBLISH_GET_PATH = "/publish/%s/%s/0/%s/%s/%s"
     PUBLISH_POST_PATH = "/publish/%s/%s/0/%s/%s"
 
-    def __init__(self, pubnub):
+    _channel: str
+    _message: any
+    _should_store: Optional[bool]
+    _use_post: Optional[bool]
+    _meta: Optional[any]
+    _replicate: Optional[bool]
+    _ptto: Optional[int]
+    _ttl: Optional[int]
+
+    def __init__(self, pubnub, channel: Optional[str] = None, message: Optional[any] = None,
+                 should_store: Optional[bool] = None, use_post: Optional[bool] = None, meta: Optional[any] = None,
+                 replicate: Optional[bool] = None, ptto: Optional[int] = None, ttl: Optional[int] = None):
         super(Publish, self).__init__(pubnub)
-        self._channel = None
-        self._message = None
-        self._should_store = None
-        self._use_post = None
-        self._meta = None
-        self._replicate = None
-        self._ptto = None
-        self._ttl = None
-
-    def channel(self, channel):
+        self._channel = channel
+        self._message = message
+        self._should_store = should_store
+        self._use_post = use_post
+        self._meta = meta
+        self._replicate = replicate
+        self._ptto = ptto
+        self._ttl = ttl
+
+    def channel(self, channel: str) -> 'Publish':
         self._channel = str(channel)
         return self
 
-    def message(self, message):
+    def message(self, message: any) -> 'Publish':
         self._message = message
         return self
 
-    def use_post(self, use_post):
+    def use_post(self, use_post: bool) -> 'Publish':
         self._use_post = bool(use_post)
         return self
 
-    def use_compression(self, compress=True):
+    def use_compression(self, compress: bool = True) -> 'Publish':
         self._use_compression = bool(compress)
         return self
 
-    def is_compressable(self):
+    def is_compressable(self) -> bool:
         return True
 
-    def should_store(self, should_store):
+    def should_store(self, should_store: bool) -> 'Publish':
         self._should_store = bool(should_store)
         return self
 
-    def meta(self, meta):
+    def meta(self, meta: any) -> 'Publish':
         self._meta = meta
         return self
 
-    def ttl(self, ttl):
+    def ttl(self, ttl: int) -> 'Publish':
         self._ttl = ttl
         return self
 

pubnub/pubnub_core.py

@@ -2,6 +2,7 @@
 import time
 from warnings import warn
 from copy import deepcopy
+from typing import Optional
 from pubnub.endpoints.entities.membership.add_memberships import AddSpaceMembers, AddUserSpaces
 from pubnub.endpoints.entities.membership.update_memberships import UpdateSpaceMembers, UpdateUserSpaces
 from pubnub.endpoints.entities.membership.fetch_memberships import FetchSpaceMemberships, FetchUserMemberships
@@ -205,8 +206,11 @@ def here_now(self):
     def where_now(self):
         return WhereNow(self)
 
-    def publish(self):
-        return Publish(self)
+    def publish(self, channel: Optional[str] = None, message: Optional[any] = None, should_store: Optional[bool] = None,
+                use_post: Optional[bool] = None, meta: Optional[any] = None, replicate: Optional[bool] = None,
+                ptto: Optional[int] = None, ttl: Optional[int] = None) -> Publish:
+        return Publish(self, channel=channel, message=message, should_store=should_store, use_post=use_post, meta=meta,
+                       replicate=replicate, ptto=ptto, ttl=ttl)
 
     def grant(self):
         warn("This method will stop working on 31th December 2024. We recommend that you use grant_token() instead.",
@@ -242,8 +246,9 @@ def history(self):
     def message_counts(self):
         return MessageCount(self)
 
-    def fire(self):
-        return Fire(self)
+    def fire(self, channel: Optional[str] = None, message: Optional[any] = None, use_post: Optional[bool] = None,
+             meta: Optional[any] = None) -> Fire:
+        return Fire(self, channel=channel, message=message, use_post=use_post, meta=meta)
 
     def signal(self):
         return Signal(self)