api: Add bindings for fetching user avatar.

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,21 +728,25 @@ 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()
         marshalled_request = {}
         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:
     """

zulip/zulip/examples/get-user-avatar

@@ -0,0 +1,30 @@
+#!/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. 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))