How to use cache filter?

user_17124806 0
Viewed 301

I'm having problem with cache filter.

The idea is to not cache the responses that contain "incomplete_result":true

This is my filter function:

import requests
import requests_cache

def phrase_filter(response: requests.models.Response)->bool:
    if '"incomplete_results":true' in response.text:
        return False
    return True

But when I test it out by this code:

requests_cache.install_cache('demo_cache',expired_after=600,filter_fn=phrase_filter)
requests_cache.clear()

url1 = 'https://raw.githubusercontent.com/KienTrann/requests-cache-testing/main/should_be_cached.txt'
url2 = 'https://raw.githubusercontent.com/KienTrann/requests-cache-testing/main/should_not_be_cached.txt'

with requests_cache.enabled():
    r = requests.get(url1) # First request
    r = requests.get(url1) # Second request
    print(f'Text from url1:\n{r.text}')
    assert r.from_cache==True
    #
    r1 = requests.get(url2) # First request
    r1 = requests.get(url2) # Second request
    print('---')
    print(f'Text from url2:\n{r1.text}')
    assert r1.from_cache==False

requests_cache.disabled()

And here is the result:

Text from url1:
abc
xyz
"incomplete_results":false

---
Text from url2:
abc
xyz
"incomplete_results":true

Traceback (most recent call last):
  File "C:\Users\ADMIN\source\repos\LearningPython\py_2\py_2.py", line 25, in <module>
    assert r1.from_cache==False
AssertionError

I don't understand why r1 was cached.

What is the problem? And how can I fix it?

Thank you for taking time to answer

browser-cache caching python
2 Answers

Patching

Looks like you were almost there! requests_cache.enabled() and disabled() are contextmanager alternatives to install_cache() and uninstall_cache(). Just pass your settings to enabled() instead of install_cache():

with requests_cache.enabled('demo_cache', expire_after=600, filter_fn=phrase_filter):
    # ... make requests

This is basically the same as:

requests_cache.install_cache('demo_cache', expire_after=600, filter_fn=phrase_filter)
# ... make requests
requests_cache.uninstall_cache()

Sessions

Personally I'd recommend using requests_cache.CachedSession instead of the patching methods, because it makes it more explicit what's being cached, and if you want to make non-cached requests, you can just use the regular requests functions. There's more info in the docs here: https://requests-cache.readthedocs.io/en/stable/user_guide/general.html

Example:

from requests import Response
from requests_cache import CachedSession

def phrase_filter(response: Response) -> bool:
    return '"incomplete_results":true' not in response.text

url1 = 'https://raw.githubusercontent.com/KienTrann/requests-cache-testing/main/should_be_cached.txt'
url2 = 'https://raw.githubusercontent.com/KienTrann/requests-cache-testing/main/should_not_be_cached.txt'
session = CachedSession('demo_cache', expire_after=600, filter_fn=phrase_filter)
session.cache.clear()

nonfiltered_response = session.get(url1)
nonfiltered_response = session.get(url1)
assert nonfiltered_response.from_cache is True

filtered_response = session.get(url2)
filtered_response = session.get(url2)
assert filtered_response.from_cache is False

Debugging

If you run into similar issues in the future where you're not sure why a response is or isn't getting cached, you can just enable debug logging:

import logging
logging.basicConfig(level='DEBUG')

And you'll get caching info for each response like this:

DEBUG:requests_cache.session: Pre-cache checks for response from https://raw.githubusercontent.com/KienTrann/requests-cache-testing/main/should_not_be_cached.txt: 
{
    'disabled cache': False,
    'disabled method': False,
    'disabled status': False,
    'disabled by filter': True,
    'disabled by headers or expiration params': False,
}

More info in the docs here: https://requests-cache.readthedocs.io/en/stable/user_guide/troubleshooting.html

user_155920550

I also tried this and cannot get it to work:

# Added by Eurico Covas
# see https://requests-cache.readthedocs.io/en/stable/user_guide/filtering.html
@staticmethod
def filter_by_error(response: requests.models.Response) -> bool:
    """Don't cache responses with ErrMsg"""
    if response is None:
       return True
    if response.ok ==False:
       return True
    if len(response.json()['GDSSDKResponse']) == 1:
        if len(response.json()['GDSSDKResponse'][0]) >= 1:
           if "ErrMsg" in response.json()['GDSSDKResponse'][0].keys():
              if response.json()['GDSSDKResponse'][0]['ErrMsg'] is not None and response.json()['GDSSDKResponse'][0]['ErrMsg'] != '':
                  return True
    return False

def __init__(self, username, password, verify=True, debug=False, request_caching_enabled=False):
    assert username is not None
    assert password is not None
    assert verify is not None
    assert debug is not None
    assert request_caching_enabled is not None
    self._username = username
    self._password = password
    self._verify = verify
    self._debug = debug
    self._request_caching_enabled=request_caching_enabled
    if self._request_caching_enabled:
        self.request_count = self.get_cached_request_count()
    if not self._verify:
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    if self._debug:
        self.enable_request_debugging()
    else:
        self.enable_error_logging()
    # cache requests for 30*24 hours = 1 month!
    if self._request_caching_enabled:
        requests_cache.install_cache('capiq_cache', backend='sqlite', expire_after=30*86400, allowable_methods=('POST',), filter_fn=self.filter_by_error)

but

    response = requests.post(self._endpoint, headers=self._headers, data=json.dumps(req),
                             auth=HTTPBasicAuth(self._username, self._password), verify=self._verify)

does not call filter_by_error() ever...

user_180179290
Related