xenofem/python-zulip-api
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

flock: Move calling requests to flock's urls to separate method.

Browse source 
Introduce `make_flock_request()` method to make requests to flock's URLs.
This commit is contained in:
novokrest 2018-06-09 02:13:30 +03:00 committed by showell
parent efdc96929f
commit 7fe37dbfe8
1 changed files with 19 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
zulip_bots/zulip_bots/bots/flock/flock.py
View file

@ -18,6 +18,19 @@ def find_recipient_id(res: str, recipient_name: str) -> str:
if recipient_name == obj['firstName']: if recipient_name == obj['firstName']:
return obj['id'] return obj['id']
# Make request to given flock URL and return a two-element tuple
# whose left-hand value contains JSON body of response (or None if request failed)
# and whose right-hand value contains an error message (or None if request succeeded)
def make_flock_request(url: str, params: Dict[str, str]) -> Tuple[Any, str]:
try:
res = requests.get(url, params=params)
return (res.json(), None)
except ConnectionError as e:
logging.exception(str(e))
error = "Uh-Oh, couldn't process the request \
right now.\nPlease try again later"
return (None, error)
# Returns two-element tuple whose left-hand value contains recipient # Returns two-element tuple whose left-hand value contains recipient
# user's ID (or None if it was not found) and right-hand value contains # user's ID (or None if it was not found) and right-hand value contains
# an error message (or None if recipient user's ID was found) # an error message (or None if recipient user's ID was found)
@ -26,16 +39,10 @@ def get_recipient_id(recipient_name: str, config: Dict[str, str]) -> Tuple[Optio
payload = { payload = {
'token': token 'token': token
} }
res, error = make_flock_request(USERS_LIST_URL, payload)
try: if res is None:
res = requests.get(USERS_LIST_URL, params=payload)
except ConnectionError as e:
logging.exception(str(e))
error = "Uh-Oh, couldn't process the request \
right now.\nPlease try again later"
return (None, error) return (None, error)
res = res.json()
recipient_id = find_recipient_id(res, recipient_name) recipient_id = find_recipient_id(res, recipient_name)
if recipient_id is None: if recipient_id is None:
error = "No user found. Make sure you typed it correctly." error = "No user found. Make sure you typed it correctly."
@ -62,15 +69,11 @@ def get_flock_response(content: str, config: Dict[str, str]) -> str:
'text': message, 'text': message,
'token': token 'token': token
} }
try: res, error = make_flock_request(SEND_MESSAGE_URL, payload)
r = requests.get(SEND_MESSAGE_URL, params=payload) if res is None:
except ConnectionError as e: return error
logging.exception(str(e))
return "Uh-Oh, couldn't process the request \
right now.\nPlease try again later"
r = r.json() if "uid" in res:
if "uid" in r:
return "Message sent." return "Message sent."
else: else:
return "Message sending failed :slightly_frowning_face:. Please try again." return "Message sending failed :slightly_frowning_face:. Please try again."