File size: 2,166 Bytes
892fa81 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """Shared HTTP session — requests with connection pooling + retries."""
from __future__ import annotations
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
_session: Optional[requests.Session] = None
def shared_session() -> requests.Session:
"""Return a process-wide shared requests.Session."""
global _session
if _session is None:
_session = _make_session()
return _session
def _make_session() -> requests.Session:
session = requests.Session()
retry = Retry(
total=2,
backoff_factor=0.3,
status_forcelist=[502, 503, 504],
allowed_methods=["GET", "POST", "HEAD"],
)
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=10, max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def fetch_html(url: str, timeout: int = 15, headers: Optional[dict] = None) -> str:
"""GET a URL and return text."""
from cores.search.user_agent import random_user_agent
h = {"User-Agent": random_user_agent()}
if headers:
h.update(headers)
resp = shared_session().get(url, headers=h, timeout=timeout)
resp.raise_for_status()
return resp.text
def fetch_bytes(url: str, timeout: int = 15, headers: Optional[dict] = None) -> bytes:
"""GET a URL and return raw bytes."""
from cores.search.user_agent import random_user_agent
h = {"User-Agent": random_user_agent()}
if headers:
h.update(headers)
resp = shared_session().get(url, headers=h, timeout=timeout, stream=True)
resp.raise_for_status()
return resp.content
def fetch_json(url: str, timeout: int = 15, headers: Optional[dict] = None,
params: Optional[dict] = None) -> dict:
"""GET a URL and return parsed JSON."""
from cores.search.user_agent import random_user_agent
h = {"User-Agent": random_user_agent(), "Accept": "application/json"}
if headers:
h.update(headers)
resp = shared_session().get(url, headers=h, timeout=timeout, params=params)
resp.raise_for_status()
return resp.json()
|