Source code for pkgcheck.addons.net
"""Various support for network checks."""
import logging
import os
import warnings
from ..checks.network import RequestError, SSLError
# suppress all urllib3 log messages and import warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore")
import requests
logging.getLogger("urllib3").propagate = False
[docs]
class Session(requests.Session):
"""Custom requests session handling timeout, concurrency, and header settings."""
def __init__(self, concurrent=None, timeout=None, user_agent=None):
super().__init__()
self.timeout = timeout
# block when urllib3 connection pool is full
concurrent = concurrent if concurrent is not None else os.cpu_count() * 5
a = requests.adapters.HTTPAdapter(pool_maxsize=concurrent, pool_block=True)
self.mount("https://", a)
self.mount("http://", a)
# spoof user agent
self.headers["User-Agent"] = user_agent
[docs]
def send(self, req, **kwargs):
# forcibly use the session timeout
kwargs["timeout"] = self.timeout
try:
with super().send(req, **kwargs) as r:
r.raise_for_status()
return r
except requests.exceptions.SSLError as e:
raise SSLError(e)
except requests.exceptions.ConnectionError as e:
raise RequestError(e, "connection failed")
except requests.exceptions.RequestException as e:
raise RequestError(e)