python必背源代码requests发送HTTP请求代码

import requests

# 1. 目标地址（仅保留基础路由）
url = "https://httpbin.org/get"

# 2. 查询参数字典（自动进行 URL 编码）
params = {
    "query": "python 教程",
    "page": 1,
    "limit": 20
}

# 3. 规范请求头（覆盖默认的 python-requests UA，防反爬拦截）
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept": "application/json"
}

# 4. 执行请求：必须显式传入 timeout=(连接超时, 读取超时)
response = requests.get(url, params=params, headers=headers, timeout=(3.05, 10))

# 5. 状态熔断：若响应为 4xx 或 5xx，直接触发 HTTPError 异常
response.raise_for_status()

# 6. 数据解析：针对 JSON 响应直接解析为字典
result_data = response.json()
print("实际请求完整 URL:", response.url)
print("响应状态码:", response.status_code)
print("解析到的查询参数:", result_data["args"])

import requests

api_url = "https://httpbin.org/post"

# 场景 A：JSON 接口调用（微服务、现代 API 必背）
json_payload = {
    "user_id": 1001,
    "action": "checkout",
    "items": ["apple", "banana"]
}
# 使用 json= 参数，requests 自动封装 Content-Type: application/json
res_json = requests.post(api_url, json=json_payload, timeout=5)
res_json.raise_for_status()

# 场景 B：传统表单提交（普通网页登录、键值对提交）
form_payload = {
    "username": "admin",
    "password": "SecretPass123"
}
# 使用 data= 参数，requests 自动封装 Content-Type: application/x-www-form-urlencoded
res_form = requests.post(api_url, data=form_payload, timeout=5)
res_form.raise_for_status()

print("JSON 接口请求头 Content-Type:", res_json.request.headers["Content-Type"])
print("表单请求头 Content-Type:", res_form.request.headers["Content-Type"])

import requests

# 推荐使用上下文管理器 with 语句，确保网络连接池在退出作用域后自动安全释放
with requests.Session() as session:
    # 1. 统一设置会话全局请求头（所有后续请求自动继承）
    session.headers.update({
        "User-Agent": "SessionClient/1.0",
        "Accept": "application/json"
    })

    # 2. 发起首次请求：模拟服务端种植 Cookie
    set_cookie_url = "https://httpbin.org/cookies/set?auth_token=K89s7xZa"
    res1 = session.get(set_cookie_url, timeout=5)
    res1.raise_for_status()

    # 3. 发起第二次请求：验证 Cookie 是否自动流转
    check_cookie_url = "https://httpbin.org/cookies"
    res2 = session.get(check_cookie_url, timeout=5)
    res2.raise_for_status()

    # 查看 Session 内部维护的 CookieJar 与返回数据
    print("当前 Session 持久化的 Cookies:", session.cookies.get_dict())
    print("服务端读取到的 Cookies:", res2.json())

import time
import requests
from requests.exceptions import (
    RequestException,
    Timeout,
    ConnectionError,
    HTTPError
)

def safe_fetch_with_retry(url, max_retries=3, timeout=(3.05, 5)):
    headers = {"User-Agent": "ReliableWorker/2.0"}

    for attempt in range(1, max_retries + 1):
        try:
            # 发起请求并注入超时控制
            res = requests.get(url, headers=headers, timeout=timeout)
            
            # 触发 4xx/5xx 校验，抛出 HTTPError 进入异常处理流
            res.raise_for_status()
            
            # 校验通过，返回成功解析的数据
            return res.json()

        except Timeout as e:
            # 超时通常为瞬时网络抖动，适合重试
            print(f"[第 {attempt} 次重试] 连接或读取超时: {e}")
            if attempt == max_retries:
                print("达到最大重试上限，宣告失败")
                raise
            time.sleep(2 ** attempt)  # 指数退避：2s, 4s, 8s...

        except HTTPError as e:
            # 针对 4xx (客户端错误) 与 5xx (服务端内部故障) 分流判断
            status = e.response.status_code
            if 400 <= status < 500:
                print(f"客户端参数或权限错误 (HTTP {status})，无需重试: {e}")
                raise
            print(f"服务端内部故障 (HTTP {status})，准备重试...")
            time.sleep(2)

        except ConnectionError as e:
            # 域名解析失败或物理断网
            print(f"网络连接无法建立: {e}")
            raise

        except RequestException as e:
            # 其它所有 requests 未预期的基类异常兜底
            print(f"未知网络异常兜底: {e}")
            raise

# 运行示例
if __name__ == "__main__":
    data = safe_fetch_with_retry("https://httpbin.org/delay/1")
    print("请求成功，数据字节数:", len(str(data)))

import requests
from requests.exceptions import RequestException

class HTTPClient:
    def __init__(self, base_url="", timeout=(3.05, 10)):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "EnterpriseClient/1.0",
            "Accept": "application/json"
        })

    def request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}" if self.base_url else endpoint
        kwargs.setdefault("timeout", self.timeout)
        try:
            res = self.session.request(method, url, **kwargs)
            res.raise_for_status()
            return res.json()
        except requests.exceptions.HTTPError as err:
            print(f"[HTTP Error {res.status_code}] {url}: {err}")
            raise
        except requests.exceptions.Timeout:
            print(f"[Timeout Error] 请求超时: {url}")
            raise
        except RequestException as err:
            print(f"[Network Error] 请求中断: {err}")
            raise

    def close(self):
        self.session.close()

# 使用示例：
# client = HTTPClient(base_url="https://httpbin.org")
# resp = client.request("GET", "/get", params={"k": "v"})
# client.close()