87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
"""# Requests utilities.
|
|
|
|
Utility for standarized usage of HTTP requests extension libraries in a modular fashion across different libraries.
|
|
|
|
Currently supports [`requests_cache`](https://requests-cache.readthedocs.io) and [`requests_ratelimiter`](https://github.com/JWCook/requests-ratelimiter).
|
|
|
|
The intention is that a libraries can automatically setup ratelimiting for
|
|
their specific use cases, as the library most likely has a better
|
|
understanding of the ratelimiting and caching behaviour of their associated websites than the user has.
|
|
"""
|
|
|
|
import datetime
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
|
|
import requests
|
|
|
|
from ._version import __version__
|
|
|
|
__all__ = ['__version__', 'initialize_session', 'setup_limiter']
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
import requests_cache
|
|
except ImportError:
|
|
requests_cache = None
|
|
logger.warning(
|
|
'Library requests_cache could not be loaded. Automatic cache configuration will not be used.',
|
|
)
|
|
|
|
|
|
try:
|
|
import requests_ratelimiter
|
|
except ImportError:
|
|
requests_ratelimiter = None
|
|
logger.warning(
|
|
'Library requests_ratelimiter could not be loaded. Automatic ratelimit configuration will not be used.',
|
|
)
|
|
|
|
|
|
CACHE_PATH_FROM_HOME = '.local/share/usagi-keiretsu/fin_data/http_cache'
|
|
|
|
|
|
def initialize_session() -> requests.Session:
|
|
"""Creates Session with or without caching depending upon current
|
|
capabilities.
|
|
"""
|
|
logger.info('Setting up session')
|
|
|
|
# Setup cache if it can be loaded.
|
|
if requests_cache:
|
|
home = pathlib.Path(os.environ['HOME'])
|
|
cache_path = home / CACHE_PATH_FROM_HOME
|
|
cache_path.mkdir(parents=True, exist_ok=True)
|
|
session = requests_cache.CachedSession(
|
|
cache_path,
|
|
expire_after=datetime.timedelta(days=1),
|
|
stale_if_error=True,
|
|
cache_control=False,
|
|
urls_expire_after={},
|
|
)
|
|
else:
|
|
session = requests.Session()
|
|
|
|
# Return initialized session.
|
|
return session
|
|
|
|
|
|
def setup_limiter(
|
|
session: requests.Session,
|
|
url_prefix: str,
|
|
expire_after: datetime.timedelta | None = None,
|
|
**limiter_args,
|
|
) -> requests.Session:
|
|
"""Initializes limiter on session for the given domain with the given
|
|
settings.
|
|
|
|
Will always return the given session itself.
|
|
"""
|
|
if requests_ratelimiter:
|
|
session.mount(url_prefix, requests_ratelimiter.LimiterAdapter(**limiter_args))
|
|
if requests_cache and expire_after:
|
|
session.settings.urls_expire_after[url_prefix] = expire_after
|
|
return session
|