api: Add bindings for fetching user avatar.

Author: PieterCK

zulip/zulip/__init__.py

@@ -46,6 +46,11 @@
 
 API_VERSTRING = "v1/"
 
+# Endpoints that respond with an HTTP 302 redirect to a resource (e.g. the
+# avatar endpoints redirect to the actual avatar image) rather than a JSON
+# body.
+REDIRECT_ENDPOINTS = ("avatar/",)
+
 # An optional parameter to `move_topic` and `update_message` actions
 # See eg. https://zulip.com/api/update-message#parameter-propagate_mode
 EditPropagateMode = Literal["change_one", "change_all", "change_later"]
@@ -583,6 +588,7 @@ def do_api_query(
         longpolling: bool = False,
         files: Optional[List[IO[Any]]] = None,
         timeout: Optional[float] = None,
+        non_api_or_json_url: bool = False,
     ) -> Dict[str, Any]:
         if files is None:
             files = []
@@ -636,6 +642,10 @@ 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
+        # Some endpoints (e.g. avatars) respond with a redirect to a resource
+        # rather than a JSON body; for those we must not follow the redirect.
+        is_redirect_endpoint = any(url.startswith(prefix) for prefix in REDIRECT_ENDPOINTS)
         while True:
             try:
                 kwarg = "params" if method == "GET" else "data"
@@ -648,8 +658,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=not is_redirect_endpoint,
                     **kwargs,
                 )
 
@@ -683,9 +694,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 +704,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 is_redirect_endpoint and res.is_redirect:
+                end_error_retry(True)
+                return {
+                    "result": "success",
+                    "msg": "",
+                    "url": res.headers.get("Location"),
+                }
+
             try:
                 json_result = res.json()
             except Exception:
@@ -716,6 +735,7 @@ def call_endpoint(
         longpolling: bool = False,
         files: Optional[List[IO[Any]]] = None,
         timeout: Optional[float] = None,
+        non_api_or_json_url: bool = False,
     ) -> Dict[str, Any]:
         if request is None:
             request = dict()
@@ -726,11 +746,12 @@ def call_endpoint(
         versioned_url = API_VERSTRING + (url if url is not None else "")
         return self.do_api_query(
             marshalled_request,
-            versioned_url,
+            url if non_api_or_json_url else versioned_url,
             method=method,
             longpolling=longpolling,
             files=files,
             timeout=timeout,
+            non_api_or_json_url=non_api_or_json_url,
         )
 
     def call_on_each_event(
@@ -1771,6 +1792,38 @@ def move_topic(
             request=request,
         )
 
+    def get_avatar_url_by_id(self, user_id: int, medium: bool = False) -> Dict[str, Any]:
+        """
+        If `medium=False`, the default size will be returned. Avatar sizes:
+            - default `100x100` px
+            - medium `500x500` px
+
+        Example usage:
+
+        >>> client.get_avatar_by_id(user_id=8, medium=True)
+        """
+        url = f"avatar/{user_id}"
+        if medium:
+            url += "/medium"
+        response = self.call_endpoint(url=url, method="GET", non_api_or_json_url=True)
+        return response["url"]
+
+    def get_avatar_url_by_email(self, email: str, medium: bool = False) -> Dict[str, Any]:
+        """
+        If `medium=False`, the default size will be returned. Avatar sizes:
+            - default `100x100` px
+            - medium `500x500` px
+
+        Example usage:
+
+        >>> client.get_avatar_by_email(email="hamlet@zulip.com", medium=True)
+        """
+        url = f"avatar/{email}"
+        if medium:
+            url += "/medium"
+        response = self.call_endpoint(url=url, method="GET", non_api_or_json_url=True)
+        return response["url"]
+
 
 class ZulipStream:
     """

zulip/zulip/examples/get-user-avatar

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+import argparse
+
+usage = """get-user-avatar --user_id=<user_id> --email=<email address> [options]
+
+Get the avatar URL for a user.
+"""
+
+import zulip
+
+parser = zulip.add_default_arguments(argparse.ArgumentParser(usage=usage))
+parser.add_argument("--user_id", required=True)
+parser.add_argument("--email", required=True)
+options = parser.parse_args()
+
+client = zulip.init_from_options(options)
+
+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))
+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))