50 行
1.3 KiB
Python
50 行
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict
|
|
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
|
|
DEFAULT_TIMEOUT = 30
|
|
DEFAULT_USER_AGENT = "websafe-intel"
|
|
|
|
|
|
def build_session() -> requests.Session:
|
|
session = requests.Session()
|
|
retry = Retry(
|
|
total=3,
|
|
connect=3,
|
|
read=3,
|
|
status=3,
|
|
backoff_factor=0.5,
|
|
allowed_methods=frozenset(["GET", "POST"]),
|
|
status_forcelist=[429, 500, 502, 503, 504],
|
|
raise_on_status=False,
|
|
)
|
|
adapter = HTTPAdapter(max_retries=retry)
|
|
session.mount("https://", adapter)
|
|
session.mount("http://", adapter)
|
|
session.headers.update({"User-Agent": DEFAULT_USER_AGENT})
|
|
return session
|
|
|
|
|
|
def request(
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
session: requests.Session | None = None,
|
|
timeout: int = DEFAULT_TIMEOUT,
|
|
**kwargs: Any,
|
|
) -> requests.Response:
|
|
client = session or build_session()
|
|
headers = dict(kwargs.pop("headers", {}) or {})
|
|
if "User-Agent" not in headers:
|
|
headers["User-Agent"] = DEFAULT_USER_AGENT
|
|
try:
|
|
return client.request(method, url, headers=headers, timeout=timeout, **kwargs)
|
|
except requests.exceptions.SSLError:
|
|
return client.request(method, url, headers=headers, timeout=timeout, verify=False, **kwargs)
|
|
|