Merge pull request #288 from PrimozGodec/fix-cache

[FIX] Addons - Do not use cache when no permission to write

orangecanvas/application/tests/test_addons_utils.py

@@ -1,6 +1,11 @@
+import os
+import stat
 import unittest
+from tempfile import mkdtemp
 
 from pkg_resources import Requirement
+from requests import Session
+from requests_cache import CachedSession
 
 from orangecanvas.application.utils.addons import (
     Available,
@@ -9,7 +14,7 @@
     installable_from_json_response,
     installable_items,
     is_updatable,
-    prettify_name,
+    prettify_name, _session,
 )
 from orangecanvas.config import Distribution
 
@@ -92,6 +97,15 @@ def test_prettify_name(self):
         self.assertEqual('Image Analytics', prettify_name('Orange3-ImageAnalytics'))
         self.assertEqual('Survival Analysis', prettify_name('Orange3-Survival-Analysis'))
 
+    def test_session(self):
+        # when permissions - use CachedSession
+        self.assertIsInstance(_session(), CachedSession)
+
+        # when no permissions - use request's Session
+        temp_dir = mkdtemp()
+        os.chmod(temp_dir, stat.S_IRUSR)
+        self.assertIsInstance(_session(temp_dir), Session)
+
 
 if __name__ == "__main__":
     unittest.main()

orangecanvas/application/utils/addons.py

@@ -9,6 +9,7 @@
 from collections import deque
 from datetime import timedelta
 from enum import Enum
+from sqlite3 import OperationalError
 from types import SimpleNamespace
 from typing import AnyStr, Callable, List, NamedTuple, Optional, Tuple, TypeVar, Union
 
@@ -296,14 +297,21 @@ def _session(cachedir=None):
     if cachedir is None:
         cachedir = QStandardPaths.writableLocation(QStandardPaths.CacheLocation)
         cachedir = os.path.join(cachedir, "networkcache")
-    session = requests_cache.CachedSession(
-        os.path.join(cachedir, "requests.sqlite"),
-        backend="sqlite",
-        cache_control=True,
-        expire_after=timedelta(days=1),
-        stale_if_error=True,
-    )
-    return session
+    try:
+        return requests_cache.CachedSession(
+            os.path.join(cachedir, "requests.sqlite"),
+            backend="sqlite",
+            cache_control=True,
+            expire_after=timedelta(days=1),
+            stale_if_error=True,
+        )
+    except OperationalError as ex:
+        # if no permission to write in dir or read cache file return uncached session
+        log.info(
+            f"Cache file creation/opening failed with: '{str(ex)}'. "
+            f"Using requests.Session instead of cached session."
+        )
+        return requests.Session()
 
 
 def optional_map(func: Callable[[A], B]) -> Callable[[Optional[A]], Optional[B]]: