Cache statsbombpy's requests in its own session instead of patching requests globally

statsbombpy/api_client.py

@@ -1,17 +1,13 @@
 import warnings
-from tempfile import mkdtemp
 
 import requests as req
-from requests_cache import install_cache
 
 import statsbombpy.entities as ents
-from statsbombpy.config import (API_VERSION_KEYS, CACHED_CALLS_SECS, HOSTNAME,
-                                VERSIONS, _VERSION)
+from statsbombpy.cache import session
+from statsbombpy.config import API_VERSION_KEYS, HOSTNAME, VERSIONS, _VERSION
 
 HEADERS = {"User-Agent": f"statsbombpy/{_VERSION}"}
 
-install_cache(mkdtemp(), backend="sqlite", expire_after=CACHED_CALLS_SECS)
-
 # session-lifetime cache; None until the first successful fetch
 _ENDPOINT_VERSIONS = None
 
@@ -33,7 +29,7 @@ def has_auth(creds):
 
 def get_resource(url: str, creds: dict) -> list:
     auth = req.auth.HTTPBasicAuth(creds["user"], creds["passwd"])
-    resp = req.get(url, auth=auth, headers=HEADERS)
+    resp = session.get(url, auth=auth, headers=HEADERS)
     if resp.status_code != 200:
         print(f"{url} -> {resp.status_code}")
         resp = []

statsbombpy/cache.py

@@ -0,0 +1,14 @@
+import os
+from tempfile import mkdtemp
+
+from requests_cache import CachedSession
+
+from statsbombpy.config import CACHED_CALLS_SECS
+
+# Cache statsbombpy's own requests only. requests_cache.install_cache() would
+# patch requests for the whole program that imports statsbombpy.
+session = CachedSession(
+    os.path.join(mkdtemp(), "statsbombpy"),
+    backend="sqlite",
+    expire_after=CACHED_CALLS_SECS,
+)

statsbombpy/public.py

@@ -1,11 +1,10 @@
-import requests as req
-
 import statsbombpy.entities as ents
 from statsbombpy.config import OPEN_DATA_PATHS
+from statsbombpy.cache import session
 
 
 def get_response(path):
-    response = req.get(path)
+    response = session.get(path)
     response.raise_for_status()
     data = response.json()
     return data

tests/statsbombpy_test/test_sb.py

@@ -316,5 +316,27 @@ def test_events_with_version_override(self):
         self.assertFalse(events.empty)
 
 
+class TestRequestCache(TestCase):
+    def test_importing_statsbombpy_leaves_requests_alone(self):
+        import requests
+        import requests_cache
+
+        self.assertFalse(requests_cache.is_installed())
+        self.assertNotIsInstance(requests.Session(), requests_cache.CachedSession)
+
+    def test_both_clients_use_the_statsbombpy_session(self):
+        import requests_cache
+        from statsbombpy import public
+        from statsbombpy.cache import session
+
+        self.assertIsInstance(session, requests_cache.CachedSession)
+        with patch.object(session, "get") as get:
+            get.return_value.status_code = 200
+            get.return_value.json.return_value = []
+            public.get_response("https://example.test/open-data.json")
+            api_client.get_resource("https://example.test/api", DEFAULT_CREDS)
+        self.assertEqual(get.call_count, 2)
+
+
 if __name__ == "__main__":
     main()