From 3f2a94b7f89308f64d8908f2989fcb35435d5673 Mon Sep 17 00:00:00 2001 From: PieterCK Date: Mon, 15 Jun 2026 20:29:00 +0700 Subject: [PATCH] api: Add bindings for fetching user avatar. This adds methods for fetching user avatars to the Client class. There is some plumbing in call_endpoint to make sure it can call endpoints that don't include the API version string (/api/v1), such as the avatar endpoints. --- zulip/zulip/__init__.py | 76 +++++++++++++++++++++++++--- zulip/zulip/examples/get-user-avatar | 30 +++++++++++ 2 files changed, 100 insertions(+), 6 deletions(-) create mode 100755 zulip/zulip/examples/get-user-avatar diff --git a/zulip/zulip/__init__.py b/zulip/zulip/__init__.py index 6d725a10b..3c4ce8278 100644 --- a/zulip/zulip/__init__.py +++ b/zulip/zulip/__init__.py @@ -583,6 +583,8 @@ def do_api_query( longpolling: bool = False, files: Optional[List[IO[Any]]] = None, timeout: Optional[float] = None, + non_api_or_json_url: bool = False, + allow_redirects: bool = True, ) -> Dict[str, Any]: if files is None: files = [] @@ -636,6 +638,7 @@ def end_error_retry(succeeded: bool) -> None: else: print("Failed!") + base_url = self.base_url.removesuffix("api/") if non_api_or_json_url else self.base_url while True: try: kwarg = "params" if method == "GET" else "data" @@ -648,8 +651,9 @@ def end_error_retry(succeeded: bool) -> None: # Actually make the request! res = self.session.request( method, - urllib.parse.urljoin(self.base_url, url), + urllib.parse.urljoin(base_url, url), timeout=request_timeout, + allow_redirects=allow_redirects, **kwargs, ) @@ -683,9 +687,7 @@ def end_error_retry(succeeded: bool) -> None: # go into retry logic, because the most likely scenario here is # that somebody just hasn't started their server, or they passed # in an invalid site. - raise UnrecoverableNetworkError( - "cannot connect to server " + self.base_url - ) from e + raise UnrecoverableNetworkError("cannot connect to server " + base_url) from e if error_retry(""): continue @@ -695,6 +697,16 @@ def end_error_retry(succeeded: bool) -> None: # We'll split this out into more cases as we encounter new bugs. raise + # Redirect endpoints have no JSON body; the resource URL lives in the + # "Location" header of the 3xx response. + if res.is_redirect: + end_error_retry(True) + return { + "result": "redirect", + "msg": "", + "url": res.headers.get("Location"), + } + try: json_result = res.json() except Exception: @@ -716,6 +728,8 @@ def call_endpoint( longpolling: bool = False, files: Optional[List[IO[Any]]] = None, timeout: Optional[float] = None, + non_api_or_json_url: bool = False, + allow_redirects: bool = True, ) -> Dict[str, Any]: if request is None: request = dict() @@ -723,14 +737,16 @@ def call_endpoint( for k, v in request.items(): if v is not None: marshalled_request[k] = v - versioned_url = API_VERSTRING + (url if url is not None else "") + url = url or "" return self.do_api_query( marshalled_request, - versioned_url, + url=url if non_api_or_json_url else API_VERSTRING + url, method=method, longpolling=longpolling, files=files, timeout=timeout, + non_api_or_json_url=non_api_or_json_url, + allow_redirects=allow_redirects, ) def call_on_each_event( @@ -1771,6 +1787,54 @@ def move_topic( request=request, ) + def _get_avatar_url(self, url: str) -> Union[str, Dict[str, Any]]: + response = self.call_endpoint( + url=url, method="GET", non_api_or_json_url=True, allow_redirects=False + ) + if response["result"] != "redirect": + return response + return response["url"] + + def get_avatar_url_by_id( + self, user_id: int, medium: bool = False + ) -> Union[str, Dict[str, Any]]: + """ + If `medium=False`, the default size will be returned. Avatar sizes: + - default `100x100` px + - medium `500x500` px + + Returns the avatar URL as a string on success, or the error + response dict otherwise. + + Example usage: + + >>> client.get_avatar_url_by_id(user_id=8, medium=True) + """ + url = f"avatar/{user_id}" + if medium: + url += "/medium" + return self._get_avatar_url(url) + + def get_avatar_url_by_email( + self, email: str, medium: bool = False + ) -> Union[str, Dict[str, Any]]: + """ + If `medium=False`, the default size will be returned. Avatar sizes: + - default `100x100` px + - medium `500x500` px + + Returns the avatar URL as a string on success, or the error + response dict otherwise. + + Example usage: + + >>> client.get_avatar_url_by_email(email="hamlet@zulip.com", medium=True) + """ + url = f"avatar/{urllib.parse.quote(email, safe='')}" + if medium: + url += "/medium" + return self._get_avatar_url(url) + class ZulipStream: """ diff --git a/zulip/zulip/examples/get-user-avatar b/zulip/zulip/examples/get-user-avatar new file mode 100755 index 000000000..0a4e72c22 --- /dev/null +++ b/zulip/zulip/examples/get-user-avatar @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 + +import argparse + +usage = """get-user-avatar [--user-id=] [--email=] [options] + +Get the avatar URL for a user. At least one of --user-id or --email +must be specified. +""" + +import zulip + +parser = zulip.add_default_arguments(argparse.ArgumentParser(usage=usage)) +parser.add_argument("--user-id", type=int) +parser.add_argument("--email") +options = parser.parse_args() + +if options.user_id is None and options.email is None: + parser.error("at least one of --user-id or --email is required") + +client = zulip.init_from_options(options) + +if options.user_id is not None: + print("Get user avatar by ID:") + print(client.get_avatar_url_by_id(user_id=options.user_id)) + print(client.get_avatar_url_by_id(user_id=options.user_id, medium=True)) +if options.email is not None: + print("Get user avatar by email:") + print(client.get_avatar_url_by_email(email=options.email)) + print(client.get_avatar_url_by_email(email=options.email, medium=True))