in firebase_admin/messaging.py [0:0]
def make_topic_management_request(self, tokens, topic, operation):
"""Invokes the IID service for topic management functionality."""
if isinstance(tokens, str):
tokens = [tokens]
if not isinstance(tokens, list) or not tokens:
raise ValueError('Tokens must be a string or a non-empty list of strings.')
invalid_str = [t for t in tokens if not isinstance(t, str) or not t]
if invalid_str:
raise ValueError('Tokens must be non-empty strings.')
if not isinstance(topic, str) or not topic:
raise ValueError('Topic must be a non-empty string.')
if not topic.startswith('/topics/'):
topic = '/topics/{0}'.format(topic)
data = {
'to': topic,
'registration_tokens': tokens,
}
url = '{0}/{1}'.format(_MessagingService.IID_URL, operation)
try:
resp = self._client.body(
'post',
url=url,
json=data,
headers=_MessagingService.IID_HEADERS
)
except requests.exceptions.RequestException as error:
raise self._handle_iid_error(error)
else:
return TopicManagementResponse(resp)