feat: cache upload handles (#766)

* sort files for uploading

* rename retryable checks

* refactor

* handle KeyboardInterrupt

* add upload_failed

* log if upload is resumed or not

* improve http error message

* improve logging

* log multiple lines

* handle keyboard interruptions

* remove read timeout

* handle json errors

* update upload_api_v4

* refactor: add SingleImageUploader

* fix logging

* add PersistentCache and tests

* use the persistent cache

* fix AttributeError: '_dbm.dbm' object has no attribute 'items'

Author: ptpt

mapillary_tools/api_v4.py

@@ -21,6 +21,17 @@
 USE_SYSTEM_CERTS: bool = False
 
 
+class HTTPContentError(Exception):
+    """
+    Raised when the HTTP response is ok (200) but the content is not as expected
+    e.g. not JSON or not a valid response.
+    """
+
+    def __init__(self, message: str, response: requests.Response):
+        self.response = response
+        super().__init__(message)
+
+
 class ClusterFileType(enum.Enum):
     ZIP = "zip"
     BLACKVUE = "mly_blackvue_video"
@@ -58,24 +69,25 @@ def cert_verify(self, *args, **kwargs):
 
 
 @T.overload
-def _truncate(s: bytes, limit: int = 512) -> bytes: ...
+def _truncate(s: bytes, limit: int = 256) -> bytes | str: ...
 
 
 @T.overload
-def _truncate(s: str, limit: int = 512) -> str: ...
+def _truncate(s: str, limit: int = 256) -> str: ...
 
 
-def _truncate(s, limit=512):
+def _truncate(s, limit=256):
     if limit < len(s):
+        if isinstance(s, bytes):
+            try:
+                s = s.decode("utf-8")
+            except UnicodeDecodeError:
+                pass
         remaining = len(s) - limit
         if isinstance(s, bytes):
-            return (
-                s[:limit]
-                + b"..."
-                + f"({remaining} more bytes truncated)".encode("utf-8")
-            )
+            return s[:limit] + f"...({remaining} bytes truncated)".encode("utf-8")
         else:
-            return str(s[:limit]) + f"...({remaining} more chars truncated)"
+            return str(s[:limit]) + f"...({remaining} chars truncated)"
     else:
         return s
 
@@ -95,7 +107,10 @@ def _sanitize(headers: T.Mapping[T.Any, T.Any]) -> T.Mapping[T.Any, T.Any]:
         ]:
             new_headers[k] = "[REDACTED]"
         else:
-            new_headers[k] = _truncate(v)
+            if isinstance(v, (str, bytes)):
+                new_headers[k] = T.cast(T.Any, _truncate(v))
+            else:
+                new_headers[k] = v
 
     return new_headers
 
@@ -106,7 +121,6 @@ def _log_debug_request(
     json: dict | None = None,
     params: dict | None = None,
     headers: dict | None = None,
-    timeout: T.Any = None,
 ):
     if logging.getLogger().getEffectiveLevel() <= logging.DEBUG:
         return
@@ -126,8 +140,7 @@ def _log_debug_request(
     if headers:
         msg += f" HEADERS={_sanitize(headers)}"
 
-    if timeout is not None:
-        msg += f" TIMEOUT={timeout}"
+    msg = msg.replace("\n", "\\n")
 
     LOG.debug(msg)
 
@@ -136,26 +149,41 @@ def _log_debug_response(resp: requests.Response):
     if logging.getLogger().getEffectiveLevel() <= logging.DEBUG:
         return
 
-    data: str | bytes
+    elapsed = resp.elapsed.total_seconds() * 1000  # Convert to milliseconds
+    msg = f"HTTP {resp.status_code} {resp.reason} ({elapsed:.0f} ms): {str(_truncate_response_content(resp))}"
+
+    LOG.debug(msg)
+
+
+def _truncate_response_content(resp: requests.Response) -> str | bytes:
     try:
-        data = _truncate(dumps(_sanitize(resp.json())))
-    except Exception:
-        data = _truncate(resp.content)
+        json_data = resp.json()
+    except requests.JSONDecodeError:
+        if resp.content is not None:
+            data = _truncate(resp.content)
+        else:
+            data = ""
+    else:
+        if isinstance(json_data, dict):
+            data = _truncate(dumps(_sanitize(json_data)))
+        else:
+            data = _truncate(str(json_data))
+
+    if isinstance(data, bytes):
+        return data.replace(b"\n", b"\\n")
 
-    LOG.debug(f"HTTP {resp.status_code} ({resp.reason}): %s", data)
+    elif isinstance(data, str):
+        return data.replace("\n", "\\n")
+
+    return data
 
 
 def readable_http_error(ex: requests.HTTPError) -> str:
-    req = ex.request
-    resp = ex.response
+    return readable_http_response(ex.response)
 
-    data: str | bytes
-    try:
-        data = _truncate(dumps(_sanitize(resp.json())))
-    except Exception:
-        data = _truncate(resp.content)
 
-    return f"{req.method} {resp.url} => {resp.status_code} ({resp.reason}): {str(data)}"
+def readable_http_response(resp: requests.Response) -> str:
+    return f"{resp.request.method} {resp.url} => {resp.status_code} {resp.reason}: {str(_truncate_response_content(resp))}"
 
 
 def request_post(
@@ -174,7 +202,6 @@ def request_post(
             json=json,
             params=kwargs.get("params"),
             headers=kwargs.get("headers"),
-            timeout=kwargs.get("timeout"),
         )
 
     if USE_SYSTEM_CERTS:
@@ -208,11 +235,7 @@ def request_get(
 
     if not disable_debug:
         _log_debug_request(
-            "GET",
-            url,
-            params=kwargs.get("params"),
-            headers=kwargs.get("headers"),
-            timeout=kwargs.get("timeout"),
+            "GET", url, params=kwargs.get("params"), headers=kwargs.get("headers")
         )
 
     if USE_SYSTEM_CERTS:
@@ -335,10 +358,7 @@ def fetch_user_or_me(
 def log_event(action_type: ActionType, properties: dict) -> requests.Response:
     resp = request_post(
         f"{MAPILLARY_GRAPH_API_ENDPOINT}/logging",
-        json={
-            "action_type": action_type,
-            "properties": properties,
-        },
+        json={"action_type": action_type, "properties": properties},
         headers={
             "Authorization": f"OAuth {MAPILLARY_CLIENT_TOKEN}",
         },
@@ -374,3 +394,13 @@ def finish_upload(
     resp.raise_for_status()
 
     return resp
+
+
+def jsonify_response(resp: requests.Response) -> T.Any:
+    """
+    Convert the response to JSON, raising HTTPContentError if the response is not JSON.
+    """
+    try:
+        return resp.json()
+    except requests.JSONDecodeError as ex:
+        raise HTTPContentError("Invalid JSON response", resp) from ex

mapillary_tools/authenticate.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import getpass
-import json
 import logging
 import re
 import sys
@@ -131,15 +130,18 @@ def fetch_user_items(
     user_items = _verify_user_auth(_validate_profile(user_items))
 
     LOG.info(
-        'Uploading to profile "%s": %s', profile_name, api_v4._sanitize(user_items)
+        f'Uploading to profile "{profile_name}": {user_items.get("MAPSettingsUsername")} (ID: {user_items.get("MAPSettingsUserKey")})'
     )
 
     if organization_key is not None:
         resp = api_v4.fetch_organization(
             user_items["user_upload_token"], organization_key
         )
-        LOG.info("Uploading to Mapillary organization: %s", json.dumps(resp.json()))
-        user_items["MAPOrganizationKey"] = organization_key
+        data = api_v4.jsonify_response(resp)
+        LOG.info(
+            f"Uploading to organization: {data.get('name')} (ID: {data.get('id')})"
+        )
+        user_items["MAPOrganizationKey"] = data.get("id")
 
     return user_items
 
@@ -182,12 +184,12 @@ def _verify_user_auth(user_items: config.UserItem) -> config.UserItem:
         else:
             raise ex
 
-    user_json = resp.json()
+    data = api_v4.jsonify_response(resp)
 
     return {
         **user_items,
-        "MAPSettingsUsername": user_json.get("username"),
-        "MAPSettingsUserKey": user_json.get("id"),
+        "MAPSettingsUsername": data.get("username"),
+        "MAPSettingsUserKey": data.get("id"),
     }
 
 
@@ -285,7 +287,7 @@ def _prompt_login(
 
         raise ex
 
-    data = resp.json()
+    data = api_v4.jsonify_response(resp)
 
     user_items: config.UserItem = {
         "user_upload_token": str(data["access_token"]),

mapillary_tools/commands/__main__.py

@@ -7,6 +7,7 @@
 import requests
 
 from .. import api_v4, constants, exceptions, VERSION
+from ..upload import log_exception
 from . import (
     authenticate,
     process,
@@ -162,14 +163,16 @@ def main():
     try:
         args.func(argvars)
     except requests.HTTPError as ex:
-        LOG.error("%s: %s", ex.__class__.__name__, api_v4.readable_http_error(ex))
+        log_exception(ex)
         # TODO: standardize exit codes as exceptions.MapillaryUserError
         sys.exit(16)
 
+    except api_v4.HTTPContentError as ex:
+        log_exception(ex)
+        sys.exit(17)
+
     except exceptions.MapillaryUserError as ex:
-        LOG.error(
-            "%s: %s", ex.__class__.__name__, ex, exc_info=log_level == logging.DEBUG
-        )
+        log_exception(ex)
         sys.exit(ex.exit_code)
 
 

mapillary_tools/constants.py

@@ -2,6 +2,7 @@
 
 import functools
 import os
+import tempfile
 
 import appdirs
 
@@ -146,6 +147,10 @@ def _parse_scaled_integers(
 MAPILLARY_UPLOAD_HISTORY_PATH: str = os.getenv(
     "MAPILLARY_UPLOAD_HISTORY_PATH", os.path.join(USER_DATA_DIR, "upload_history")
 )
+UPLOAD_CACHE_DIR: str = os.getenv(
+    _ENV_PREFIX + "UPLOAD_CACHE_DIR",
+    os.path.join(tempfile.gettempdir(), "mapillary_tools", "upload_cache"),
+)
 MAX_IMAGE_UPLOAD_WORKERS: int = int(
     os.getenv(_ENV_PREFIX + "MAX_IMAGE_UPLOAD_WORKERS", 64)
 )