1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
from pubnub import utils
from pubnub.endpoints.endpoint import Endpoint
from pubnub.enums import HttpMethod, PNOperationType, PNPushEnvironment, PNPushType
from pubnub.errors import (
PNERR_PUSH_DEVICE_MISSING,
PNERR_PUSH_TOPIC_MISSING,
PNERROR_PUSH_TYPE_MISSING,
)
from pubnub.exceptions import PubNubException
from pubnub.models.consumer.push import PNPushRemoveAllChannelsResult
class RemoveDeviceFromPush(Endpoint):
# v1/push/sub-key/{subKey}/devices/{pushToken}/remove
REMOVE_PATH = "/v1/push/sub-key/%s/devices/%s/remove"
# v2/push/sub-key/{subKey}/devices-apns2/{deviceApns2}/remove
REMOVE_PATH_APNS2 = "/v2/push/sub-key/%s/devices-apns2/%s/remove"
def __init__(self, pubnub):
Endpoint.__init__(self, pubnub)
self._device_id = None
self._push_type = None
self._topic = None
self._environment = None
def device_id(self, device_id):
self._device_id = device_id
return self
def push_type(self, push_type):
self._push_type = push_type
return self
def topic(self, topic):
self._topic = topic
return self
def environment(self, environment):
self._environment = environment
return self
def custom_params(self):
params = {}
if self._push_type != PNPushType.APNS2:
params["type"] = utils.push_type_to_string(self._push_type)
else:
if self._environment is None:
self._environment = PNPushEnvironment.DEVELOPMENT
params["environment"] = self._environment
params["topic"] = self._topic
return params
def build_path(self):
if self._push_type != PNPushType.APNS2:
return RemoveDeviceFromPush.REMOVE_PATH % (
self.pubnub.config.subscribe_key,
self._device_id,
)
else:
return RemoveDeviceFromPush.REMOVE_PATH_APNS2 % (
self.pubnub.config.subscribe_key,
self._device_id,
)
def http_method(self):
return HttpMethod.GET
def validate_params(self):
self.validate_subscribe_key()
if not isinstance(self._device_id, str) or len(self._device_id) == 0:
raise PubNubException(pn_error=PNERR_PUSH_DEVICE_MISSING)
if self._push_type is None:
raise PubNubException(pn_error=PNERROR_PUSH_TYPE_MISSING)
if self._push_type == PNPushType.APNS2:
if not isinstance(self._topic, str) or len(self._topic) == 0:
raise PubNubException(pn_error=PNERR_PUSH_TOPIC_MISSING)
def create_response(self, envelope):
return PNPushRemoveAllChannelsResult()
def is_auth_required(self):
return True
def request_timeout(self):
return self.pubnub.config.non_subscribe_request_timeout
def connect_timeout(self):
return self.pubnub.config.connect_timeout
def operation_type(self):
return PNOperationType.PNRemoveAllPushNotificationsOperation
def name(self):
return "RemoveDeviceFromPush"
|
