Feature/replace caching

.github/workflows/pytest.yml

@@ -32,6 +32,18 @@ jobs:
       run: |
         pip install ruff
         ruff format --check pori_python tests
+    - name: short tests
+      run: |
+        pip list
+        pytest --junitxml=junit/test-results-${{ matrix.python-version }}.xml --cov pori_python --cov-report term --cov-report xml
+      env:
+        IPR_USER: ${{ secrets.IPR_TEST_USER }}
+        IPR_PASS: ${{ secrets.IPR_TEST_PASSWORD }}
+        GRAPHKB_USER: ${{ secrets.GKB_TEST_USER }}
+        GRAPHKB_PASS: ${{ secrets.GKB_TEST_PASS }}
+        GRAPHKB_URL: ${{ secrets.GKB_TEST_URL }}
+        EXCLUDE_INTEGRATION_TESTS: 1
+      if: matrix.python-version != 3.11
     - name: Full Tests with pytest
       run: |
         pip list
@@ -42,16 +54,14 @@ jobs:
         GRAPHKB_USER: ${{ secrets.GKB_TEST_USER }}
         GRAPHKB_PASS: ${{ secrets.GKB_TEST_PASS }}
         GRAPHKB_URL: ${{ secrets.GKB_TEST_URL }}
-        # SDEV-3381 - Turn off integration tests temporarily, till efficiency is increased
-        # turn on integration tests for one python version only
-        EXCLUDE_INTEGRATION_TESTS: ${{ matrix.python-version != '3.11' }}
+      if: matrix.python-version == 3.11
     - name: Upload pytest test results
       uses: actions/upload-artifact@master
       with:
         name: pytest-results-${{ matrix.python-version }}
         path: junit/test-results-${{ matrix.python-version }}.xml
         # Use always() to always run this step to publish test results when there are test failures
-      if: matrix.python-version == 3.9
+      if: matrix.python-version == 3.11
     - name: Update code coverage report to CodeCov
       uses: codecov/codecov-action@v3
       with:
@@ -61,4 +71,4 @@ jobs:
         env_vars: OS,PYTHON
         name: codecov-umbrella
         fail_ci_if_error: true
-      if: matrix.python-version == 3.9
+      if: matrix.python-version == 3.11

.gitignore

@@ -15,6 +15,7 @@ htmlcov
 # common virtual environment names
 venv*
 env
+.venv
 
 # editors
 .idea

pori_python/graphkb/match.py

@@ -31,12 +31,7 @@
     looks_like_rid,
     stringifyVariant,
 )
-from .vocab import (
-    get_equivalent_terms,
-    get_term_by_name,
-    get_term_tree,
-    get_terms_set,
-)
+from .vocab import get_equivalent_terms, get_term_by_name, get_term_tree, get_terms_set
 
 FEATURES_CACHE: Set[str] = set()
 
@@ -111,22 +106,6 @@ def get_equivalent_features(
     )
 
 
-def cache_missing_features(conn: GraphKBConnection) -> None:
-    """
-    Create a cache of features that exist to avoid repeatedly querying
-    for missing features
-    """
-    genes = cast(
-        List[Ontology],
-        conn.query({'target': 'Feature', 'returnProperties': ['name', 'sourceId'], 'neighbors': 0}),
-    )
-    for gene in genes:
-        if gene['name']:
-            FEATURES_CACHE.add(gene['name'].lower())
-        if gene['sourceId']:
-            FEATURES_CACHE.add(gene['sourceId'].lower())
-
-
 def match_category_variant(
     conn: GraphKBConnection,
     reference_name: str,

pori_python/graphkb/util.py

@@ -1,27 +1,27 @@
-import requests
-from requests.adapters import HTTPAdapter
-
-import hashlib
 import json
 import logging
 import os
 import re
 import time
 from datetime import datetime
-from typing import Any, Dict, Iterable, List, Optional, Union, cast
-from urllib3.util.retry import Retry
+from typing import Dict, Iterable, List, Optional, Union, cast
 from urllib.parse import urlsplit
 
+import requests
+from requests.adapters import HTTPAdapter
+from requests_cache import CachedSession
+from requests_ratelimiter import LimiterAdapter
+from urllib3.util.retry import Retry
+
 from pori_python.types import ParsedVariant, PositionalVariant, Record
 
 from .constants import DEFAULT_LIMIT, TYPES_TO_NOTATION, AA_3to1_MAPPING
 
-QUERY_CACHE: Dict[Any, Any] = {}
-
 # name the logger after the package to make it simple to disable for packages using this one as a dependency
 # https://stackoverflow.com/questions/11029717/how-do-i-disable-log-messages-from-the-requests-library
 
 logger = logging.getLogger('graphkb')
+DEFAULT_LIMITER = LimiterAdapter(per_second=3)
 
 
 def convert_to_rid_list(records: Iterable[Record]) -> List[str]:
@@ -88,37 +88,91 @@ def millis_interval(start: datetime, end: datetime) -> int:
     return millis
 
 
-def cache_key(request_body) -> str:
-    """Create a cache key for a query request to GraphKB."""
-    body = json.dumps(request_body, sort_keys=True)
-    hash_code = hashlib.md5(f'/query{body}'.encode('utf-8')).hexdigest()
-    return hash_code
-
-
 class GraphKBConnection:
     def __init__(
         self,
         url: str = os.environ.get('GRAPHKB_URL'),
         username: str = '',
         password: str = '',
         use_global_cache: bool = True,
+        cache_name: str = '',
+        only_if_cached: bool = False,
+        session: Optional[requests.Session] = None,
+        limiter: LimiterAdapter = DEFAULT_LIMITER,
+        **session_kwargs,
     ):
-        self.http = requests.Session()
-        retries = Retry(
-            total=100,
-            connect=5,
-            status=5,
-            backoff_factor=5,
-            status_forcelist=[429, 500, 502, 503, 504],
-        )
-        self.http.mount('https://', HTTPAdapter(max_retries=retries))
+        """
+        Docstring for __init__
+
+        Args:
+        - use_global_cache: cache requests across all requests to GKB
+        - cache_name: Path or connection URL to the database which stors the requests cache. see https://requests-cache.readthedocs.io/en/v0.6.4/user_guide.html#cache-name
+        - only_if_cached: this will set the cache-control header for all requests to only-if-cached which will raise 504 errors if a request does not exist in the cache already rather than making a new network request
+        """
+        session_cls = requests.Session
+        if limiter and not use_global_cache:
+            raise NotImplementedError(f'currently rate limiting by default also implements caching')
+        if session is not None:
+            if limiter is not None:
+                raise NotImplementedError('cannot add limiter to an existing session')
+            if use_global_cache:
+                raise NotImplementedError(
+                    'the use_global_cache parameter should not be used with a custom session'
+                )
+            if cache_name:
+                raise NotImplementedError(
+                    'cache_name should not be used with a custom input session'
+                )
+        if not use_global_cache and cache_name:
+            raise NotImplementedError('cache_name only applies when use_global_cache is True')
+
+        if use_global_cache:
+            session_cls = CachedSession
+            if not cache_name:
+                session_kwargs['backend'] = 'memory'
+            else:
+                session_kwargs['cache_name'] = cache_name
+            session_kwargs['allowable_methods'] = ['GET', 'POST']
+            session_kwargs['ignored_parameters'] = ['Authorization']
+            session_kwargs['cache_control'] = True
+
+        if 'PYTEST_CURRENT_TEST' in os.environ or only_if_cached:
+            logging.warning(
+                f'rate limiting is by default turned off for tests and cache-only queries'
+            )
+            limiter = None
+
+        if not session:
+            self.http = session_cls(**session_kwargs)
+        else:
+            self.http = session
+
+        if limiter is not None:
+            self.http.mount('http://', limiter)
+            self.http.mount('https://', limiter)
+
+        if not only_if_cached:
+            # requests-cache returns 504 when something is not in cache, since we don't want to fetch networkx requests when this flag is set, retries are redundant
+            retries = Retry(
+                total=100,
+                connect=5,
+                status=5,
+                backoff_factor=5,
+                status_forcelist=[429, 500, 502, 503, 504],
+            )
+            self.http.mount('http://', HTTPAdapter(max_retries=retries))
+            self.http.mount('https://', HTTPAdapter(max_retries=retries))
+        self.only_if_cached = only_if_cached
+
         self.token = ''
         self.token_kc = ''
         self.url = url
         self.username = username
         self.password = password
-        self.headers = {'Accept': 'application/json', 'Content-Type': 'application/json'}
-        self.cache: Dict[Any, Any] = {} if not use_global_cache else QUERY_CACHE
+        self.headers = {
+            'Accept': 'application/json',
+            'Content-Type': 'application/json',
+        }
         self.request_count = 0
         self.first_request: Optional[datetime] = None
         self.last_request: Optional[datetime] = None
@@ -137,7 +191,16 @@ def load(self) -> Optional[float]:
                 return self.request_count * 1000 / msec
         return None
 
-    def request(self, endpoint: str, method: str = 'GET', **kwargs) -> Dict:
+    def request(
+        self,
+        endpoint: str,
+        method: str = 'GET',
+        headers: Optional[dict[str, str]] = None,
+        ignore_cache=False,
+        force_refresh=False,
+        only_if_cached=False,
+        **kwargs,
+    ) -> Dict:
         """Request wrapper to handle adding common headers and logging.
 
         Args:
@@ -147,6 +210,13 @@ def request(self, endpoint: str, method: str = 'GET', **kwargs) -> Dict:
         Returns:
             dict: the json response as a python dict
         """
+        if headers is None:
+            headers = {}
+
+        if ignore_cache or force_refresh:
+            headers['Cache-Control'] = 'no-cache'
+        elif only_if_cached or self.only_if_cached:
+            headers['Cache-Control'] = 'only-if-cached'
         url = join_url(self.url, endpoint)
         self.request_count += 1
         connect_timeout = 7
@@ -158,6 +228,11 @@ def request(self, endpoint: str, method: str = 'GET', **kwargs) -> Dict:
         if endpoint in ['query', 'parse']:
             timeout = (connect_timeout, read_timeout)
 
+        request_headers = {}
+        request_headers.update(self.headers)
+        if headers is not None:
+            request_headers.update(headers)
+
         start_time = datetime.now()
 
         if not self.first_request:
@@ -179,8 +254,8 @@ def request(self, endpoint: str, method: str = 'GET', **kwargs) -> Dict:
                     need_refresh_login = False
 
                 self.request_count += 1
-                resp = requests.request(
-                    method, url, headers=self.headers, timeout=timeout, **kwargs
+                resp = self.http.request(
+                    method, url, headers=request_headers, timeout=timeout, **kwargs
                 )
                 if resp.status_code == 401 or resp.status_code == 403:
                     logger.debug(f'/{endpoint} - {resp.status_code} - retrying')
@@ -293,39 +368,29 @@ def login(self, username: str, password: str, pori_demo: bool = False) -> None:
     def refresh_login(self) -> None:
         self.login(self.username, self.password)
 
-    def set_cache_data(self, request_body: Dict, result: List[Record]) -> None:
-        """Explicitly add a query to the cache."""
-        hash_code = cache_key(request_body)
-        self.cache[hash_code] = result
-
     def query(
         self,
         request_body: Dict = {},
         paginate: bool = True,
-        ignore_cache: bool = False,
-        force_refresh: bool = False,
         limit: int = DEFAULT_LIMIT,
+        **kwargs,
     ) -> List[Record]:
         """
         Query GraphKB
         """
-        result: List[Record] = []
-        hash_code = ''
-
-        if not ignore_cache and paginate:
-            hash_code = cache_key(request_body)
-            if hash_code in self.cache and not force_refresh:
-                return self.cache[hash_code]
 
+        result: List[Record] = []
         while True:
-            content = self.post('query', data={**request_body, 'limit': limit, 'skip': len(result)})
+            content = self.post(
+                'query',
+                data={**request_body, 'limit': limit, 'skip': len(result)},
+                **kwargs,
+            )
             records = content['result']
             result.extend(records)
             if len(records) < limit or not paginate:
                 break
 
-        if not ignore_cache and paginate:
-            self.cache[hash_code] = result
         return result
 
     def parse(self, hgvs_string: str, requireFeatures: bool = False) -> ParsedVariant:
@@ -500,7 +565,9 @@ def stripDisplayName(displayName: str, withRef: bool = True, withRefSeq: bool =
 
 
 def stringifyVariant(
-    variant: Union[PositionalVariant, ParsedVariant], withRef: bool = True, withRefSeq: bool = True
+    variant: Union[PositionalVariant, ParsedVariant],
+    withRef: bool = True,
+    withRefSeq: bool = True,
 ) -> str:
     """
     Convert variant record to a string representation (displayName/hgvs)

pori_python/graphkb/vocab.py

@@ -181,9 +181,6 @@ def get_terms_set(
 ) -> Set[str]:
     """Get a set of vocabulary rids given some base/parent term names."""
     base_terms = [base_terms] if isinstance(base_terms, str) else base_terms
-    cache_key = tuple(sorted(base_terms))
-    if graphkb_conn.cache.get(cache_key, None) and not ignore_cache:
-        return graphkb_conn.cache[cache_key]
     terms = set()
     for base_term in base_terms:
         terms.update(
@@ -193,6 +190,4 @@ def get_terms_set(
                 )
             )
         )
-    if not ignore_cache:
-        graphkb_conn.cache[cache_key] = terms
     return terms

pori_python/ipr/main.py

@@ -33,8 +33,8 @@
     preprocess_cosmic,
     preprocess_expression_variants,
     preprocess_hla,
-    preprocess_msi,
     preprocess_hrd,
+    preprocess_msi,
     preprocess_signature_variants,
     preprocess_small_mutations,
     preprocess_structural_variants,

setup.cfg

@@ -36,6 +36,9 @@ install_requires =
     requests
     tqdm
     typing_extensions>=3.7.4.2,<5
+    requests-cache[sqlite]
+    requests-ratelimiter
+    pyrate-limiter
 
 [options.extras_require]
 deploy = twine; wheel; m2r
@@ -62,3 +65,8 @@ pori_python = py.typed
 [options.entry_points]
 console_scripts =
     ipr = pori_python.ipr.main:command_interface
+
+
+[tool:pytest]
+log_cli = true
+log_cli_level = INFO