feat: improve authentication

mapillary_tools/api_v4.py

@@ -70,7 +70,7 @@ def _truncate(s, limit=512):
         return s
 
 
-def _sanitize(headers: T.Dict):
+def _sanitize(headers: T.Mapping[T.Any, T.Any]) -> T.Mapping[T.Any, T.Any]:
     new_headers = {}
 
     for k, v in headers.items():
@@ -81,6 +81,7 @@ def _sanitize(headers: T.Dict):
             "access-token",
             "access_token",
             "password",
+            "user_upload_token",
         ]:
             new_headers[k] = "[REDACTED]"
         else:
@@ -224,6 +225,44 @@ def request_get(
     return resp
 
 
+def is_auth_error(resp: requests.Response) -> bool:
+    if resp.status_code in [401, 403]:
+        return True
+
+    if resp.status_code in [400]:
+        try:
+            error_body = resp.json()
+        except Exception:
+            error_body = {}
+
+        type = error_body.get("debug_info", {}).get("type")
+        if type in ["NotAuthorizedError"]:
+            return True
+
+    return False
+
+
+def extract_auth_error_message(resp: requests.Response) -> str:
+    assert is_auth_error(resp), "has to be an auth error"
+
+    try:
+        error_body = resp.json()
+    except Exception:
+        error_body = {}
+
+    # from Graph APIs
+    message = error_body.get("error", {}).get("message")
+    if message is not None:
+        return str(message)
+
+    # from upload service
+    message = error_body.get("debug_info", {}).get("message")
+    if message is not None:
+        return str(message)
+
+    return resp.text
+
+
 def get_upload_token(email: str, password: str) -> requests.Response:
     resp = request_post(
         f"{MAPILLARY_GRAPH_API_ENDPOINT}/login",
@@ -252,6 +291,30 @@ def fetch_organization(
     return resp
 
 
+def fetch_user_or_me(
+    user_access_token: str,
+    user_id: T.Optional[T.Union[int, str]] = None,
+) -> requests.Response:
+    if user_id is None:
+        url = f"{MAPILLARY_GRAPH_API_ENDPOINT}/me"
+    else:
+        url = f"{MAPILLARY_GRAPH_API_ENDPOINT}/{user_id}"
+
+    resp = request_get(
+        url,
+        params={
+            "fields": ",".join(["id", "username"]),
+        },
+        headers={
+            "Authorization": f"OAuth {user_access_token}",
+        },
+        timeout=REQUESTS_TIMEOUT,
+    )
+
+    resp.raise_for_status()
+    return resp
+
+
 ActionType = T.Literal[
     "upload_started_upload", "upload_finished_upload", "upload_failed_upload"
 ]