正文

HTTP协议网络编程实例从抓网页到写API详解Python requests库并发请求超时错误处理实战代码

/0 浏览量
0823

HTTP协议网络编程实例从抓网页到写API详解Python requests库并发请求超时错误处理实战代码

嘿，兄弟！今天咱们不整那些虚头巴脑的概念堆砌，直接上干货。我花了三年时间，踩过无数的坑，最后把HTTP网络编程这条路走得明明白白。这篇文章我会从头到尾带你走一遍，从最简单的抓网页，到并发请求，到处理各种超时和错误，再到自己写一个API服务。你看完之后，HTTP这块你基本可以横着走了。

先聊聊HTTP到底是个啥

你天天上网，刷微博、看视频、点外卖，但你有没有想过，这些信息到底是怎么从一个地方跑到你屏幕上的？简单说，HTTP就是一套”协议”，就像你和快递小哥之间的暗号——你告诉他你要啥，他告诉你寄到哪，他给你发货，你签收。

HTTP是请求-响应模型。你发起一个请求（Request），服务器给你一个响应（Response）。就这么简单。

举个例子，你打开浏览器输入www.baidu.com，发生了什么？

你的电脑：嘿服务器，我要看百度的首页，GET方法，HTTP/1.1版本，我的浏览器是Chrome 120
服务器：好的，这是我的回复，状态码200（表示成功），内容类型是text/html，下面是网页内容
你的电脑：收到，渲染页面

就这么几行交换，你就看到了百度首页。但背后发生的事情远比这复杂，包括DNS解析、TCP握手、SSL加密等等。不过今天咱们不讲那么深，咱们直接上手写代码。

安装requests库——别用urllib了

说实话，Python自带的urllib也能发HTTP请求，但那代码写得让人想哭。我强烈推荐requests库，它简直是好用到飞起。

pip install requests

就这一行命令，搞定。requests的哲学就是”HTTP for humans”，用起来非常直观。

第一个实战：抓取一个网页

咱们先从最简单的开始。假设你想抓取某个网页的内容，比如我帮你抓一下某个开源项目的主页。

import requests

# 最简单的GET请求
url = "https://api.github.com"
response = requests.get(url)

# 看看返回状态码
print(f"状态码: {response.status_code}")

# 看看响应内容
print(f"响应内容长度: {len(response.text)}")
print(f"前200个字符: {response.text[:200]}")

运行这段代码，你会看到类似这样的输出：

状态码: 200
响应内容长度: 1456
前200个字符: {"message":"GitHub API v2","documentation_url":"https://docs.github.com/v2"}

看到没，requests.get()返回的是一个Response对象，里面有status_code（状态码）、text（文本内容）、json()（解析为JSON）、headers（响应头）等等。

状态码是什么意思？ 这是HTTP里非常重要的概念。200表示成功，404表示找不到页面，500表示服务器内部错误，403表示没有权限，301表示重定向。这些码你以后会天天见到。

进阶：带参数的请求

很多时候你不能只发一个裸请求，需要带参数。比如搜索”python requests”，URL大概是这样的：

import requests

# 带查询参数的请求
base_url = "https://api.github.com/search/repositories"
params = {
    "q": "requests",
    "sort": "stars",
    "order": "desc",
    "per_page": 10
}

response = requests.get(base_url, params=params)

# 看看实际拼接的URL
print(f"实际请求URL: {response.url}")

# 获取结果
if response.status_code == 200:
    data = response.json()
    for item in data["items"][:5]:
        print(f"仓库: {item['full_name']}, 星数: {item['stargazers_count']}")
else:
    print(f"请求失败: {response.status_code}")

你会发现requests会自动帮你把params字典拼接到URL上，而且处理得很好。输出大概是这样：

实际请求URL: https://api.github.com/search/repositories?q=requests&sort=stars&order=desc&per_page=10
仓库: requests/requests, 星数: 45000+
仓库: psf/requests, 星数: 1200+
...

POST请求——提交数据

GET是获取数据，POST是提交数据。比如你要登录一个系统，或者提交一个表单：

import requests

# POST请求登录
login_url = "https://httpbin.org/post"

payload = {
    "username": "myuser",
    "password": "mypassword123",
    "remember_me": True
}

response = requests.post(login_url, data=payload)

if response.status_code == 200:
    data = response.json()
    print(f"提交的数据: {data['form']}")
    print(f"提交的JSON: {data['json']}")

httpbin.org是一个专门用来测试HTTP请求的网站，你发的任何请求它都会原样返回给你，非常适合学习。

带Headers的请求

有时候你需要模拟特定的浏览器，或者带上特定的认证信息：

import requests

# 模拟浏览器请求
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Authorization": "Bearer YOUR_TOKEN_HERE"
}

response = requests.get("https://httpbin.org/headers", headers=headers)
print(response.json())

很多网站会检查User-Agent，如果你不带，可能会返回403。另外Bearer Token这种认证方式现在非常普遍，你在各个平台的API里都会见到。

文件上传

import requests

# 上传文件
url = "https://httpbin.org/post"
files = {
    "file": open("important_document.pdf", "rb")
}

response = requests.post(url, files=files)
print(response.json())

上传文件用的是files参数，注意要用”rb”模式打开文件。

超时处理——这是很多人忽略的重点

兄弟，这里我要特别强调一下。很多新手写的爬虫或者网络程序，从来不管超时，结果有时候请求卡死在那里，整个程序就等着了。超时处理是生产环境必备的技能。

import requests

# 设置超时
try:
    # timeout可以是秒数，也可以是元组(连接超时, 读取超时)
    response = requests.get(
        "https://api.github.com",
        timeout=(5, 10)  # 5秒内建立连接，10秒内读取内容
    )
    print(f"请求成功，耗时: {response.elapsed.total_seconds():.2f}秒")
except requests.exceptions.Timeout:
    print("请求超时了！")
except requests.exceptions.ConnectionError:
    print("网络连接错误！")
except requests.exceptions.HTTPError as e:
    print(f"HTTP错误: {e}")
except requests.exceptions.RequestException as e:
    print(f"请求异常: {e}")

timeout参数非常关键。它接受一个元组，第一个值是连接超时（和服务器建立TCP连接的最长时间），第二个值是读取超时（等待服务器响应的最长时间）。你可以根据实际情况调整这两个值。

为什么要分开设置？ 因为有时候连接很快，但服务器响应很慢，或者网络很慢。分开设置可以更精准地控制。

重试机制——网络不稳定的救星

网络不稳定的时候，请求失败是常事。这时候你需要重试机制：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 创建一个session
session = requests.Session()

# 配置重试策略
retry_strategy = Retry(
    total=3,                    # 最多重试3次
    backoff_factor=1,           # 重试间隔: 1秒, 2秒, 4秒...
    status_forcelist=[429, 500, 502, 503, 504],  # 这些状态码才重试
    allowed_methods=["GET", "POST"]  # 只重试这些方法
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)

# 使用session发送请求
response = session.get("https://api.github.com", timeout=10)
print(f"状态码: {response.status_code}")

这个配置的意思是：如果请求返回429（太多请求）、500/502/503/504（服务器错误），就自动重试。重试间隔按照1秒、2秒、4秒递增，最多重试3次。这在处理限流或者网络波动时非常有用。

并发请求——速度提升的关键

单线程一个接一个发请求，太慢了！假设你要抓取100个网页，串行请求可能要几分钟甚至更久。这时候你需要并发。

方式一：多线程

import requests
from concurrent.futures import ThreadPoolExecutor
import time

urls = [
    "https://api.github.com/repos/requests/requests",
    "https://api.github.com/repos/python/cpython",
    "https://api.github.com/repos/torvalds/linux",
    "https://api.github.com/repos/microsoft/vscode",
    "https://api.github.com/repos/fabricjs/fabric.js",
]

def fetch_url(url):
    try:
        response = requests.get(url, timeout=10)
        return {
            "url": url,
            "status_code": response.status_code,
            "content_length": len(response.text),
            "success": True
        }
    except Exception as e:
        return {
            "url": url,
            "error": str(e),
            "success": False
        }

start_time = time.time()

# 使用线程池并发请求
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch_url, urls))

elapsed = time.time() - start_time

for result in results:
    if result["success"]:
        print(f"✅ {result['url']}: {result['status_code']}, 内容长度: {result['content_length']}")
    else:
        print(f"❌ {result['url']}: {result['error']}")

print(f"\n总共耗时: {elapsed:.2f}秒")

方式二：aiohttp异步请求（性能更好）

如果你追求更高的性能，可以用aiohttp，它是基于asyncio的异步HTTP客户端：

import asyncio
import aiohttp
import time

urls = [
    "https://api.github.com/repos/requests/requests",
    "https://api.github.com/repos/python/cpython",
    "https://api.github.com/repos/torvalds/linux",
    "https://api.github.com/repos/microsoft/vscode",
    "https://api.github.com/repos/fabricjs/fabric.js",
]

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
            content = await response.text()
            return {
                "url": url,
                "status_code": response.status,
                "content_length": len(content),
                "success": True
            }
    except Exception as e:
        return {
            "url": url,
            "error": str(e),
            "success": False
        }

async def main():
    start_time = time.time()
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    elapsed = time.time() - start_time
    
    for result in results:
        if result["success"]:
            print(f"✅ {result['url']}: {result['status_code']}, 内容长度: {result['content_length']}")
        else:
            print(f"❌ {result['url']}: {result['error']}")
    
    print(f"\n总共耗时: {elapsed:.2f}秒")

if __name__ == "__main__":
    asyncio.run(main())

对比一下，多线程方式简单直观，适合大多数场景。aiohttp性能更好，但需要理解异步编程的概念。你可以根据实际情况选择。

写一个自己的API服务

光会请求别人不够，你也要会写API。这里我用Flask来演示，它是Python最流行的轻量级Web框架：

pip install flask

基础API

from flask import Flask, jsonify, request

app = Flask(__name__)

# 模拟数据存储
users = [
    {"id": 1, "name": "张三", "age": 25, "email": "zhangsan@example.com"},
    {"id": 2, "name": "李四", "age": 30, "email": "lisi@example.com"},
    {"id": 3, "name": "王五", "age": 28, "email": "wangwu@example.com"},
]

@app.route("/api/users", methods=["GET"])
def get_users():
    """获取所有用户"""
    return jsonify({"users": users, "count": len(users)})

@app.route("/api/users/<int:user_id>", methods=["GET"])
def get_user(user_id):
    """获取单个用户"""
    user = next((u for u in users if u["id"] == user_id), None)
    if user:
        return jsonify(user)
    return jsonify({"error": "用户不存在"}), 404

@app.route("/api/users", methods=["POST"])
def create_user():
    """创建用户"""
    data = request.get_json()
    
    if not data or "name" not in data:
        return jsonify({"error": "name字段是必需的"}), 400
    
    new_user = {
        "id": max(u["id"] for u in users) + 1 if users else 1,
        "name": data["name"],
        "age": data.get("age", 0),
        "email": data.get("email", "")
    }
    users.append(new_user)
    
    return jsonify(new_user), 201

@app.route("/api/users/<int:user_id>", methods=["PUT"])
def update_user(user_id):
    """更新用户"""
    user = next((u for u in users if u["id"] == user_id), None)
    if not user:
        return jsonify({"error": "用户不存在"}), 404
    
    data = request.get_json()
    if "name" in data:
        user["name"] = data["name"]
    if "age" in data:
        user["age"] = data["age"]
    if "email" in data:
        user["email"] = data["email"]
    
    return jsonify(user)

@app.route("/api/users/<int:user_id>", methods=["DELETE"])
def delete_user(user_id):
    """删除用户"""
    global users
    users = [u for u in users if u["id"] != user_id]
    return jsonify({"message": "用户已删除"})

if __name__ == "__main__":
    app.run(debug=True, port=5000)

运行这个脚本后，你的API服务就在http://localhost:5000跑起来了。你可以用curl或者requests来测试：

import requests

BASE_URL = "http://localhost:5000/api"

# 获取所有用户
response = requests.get(f"{BASE_URL}/users")
print("获取用户列表:", response.json())

# 获取单个用户
response = requests.get(f"{BASE_URL}/users/1")
print("获取用户1:", response.json())

# 创建用户
new_user = {"name": "赵六", "age": 35, "email": "zhaoliu@example.com"}
response = requests.post(f"{BASE_URL}/users", json=new_user)
print("创建用户:", response.json())

# 更新用户
updated_user = {"age": 36, "email": "newemail@example.com"}
response = requests.put(f"{BASE_URL}/users/4", json=updated_user)
print("更新用户:", response.json())

# 删除用户
response = requests.delete(f"{BASE_URL}/users/4")
print("删除用户:", response.json())

进阶：带认证和限流的API

真实的API不可能这么裸，你还需要认证、限流、日志等功能：

from flask import Flask, jsonify, request, abort
import hashlib
import time
from functools import wraps

app = Flask(__name__)

# 简单的内存存储
users = [
    {"id": 1, "name": "张三", "age": 25, "password_hash": "abc123"},
    {"id": 2, "name": "李四", "age": 30, "password_hash": "def456"},
]

# 访问令牌存储
tokens = {}

# 请求限流
request_counts = {}

def generate_token(username, password):
    """生成简单的访问令牌"""
    token = hashlib.sha256(f"{username}{password}{time.time()}".encode()).hexdigest()
    tokens[token] = {
        "username": username,
        "created_at": time.time()
    }
    return token

def require_auth(f):
    """认证装饰器"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            abort(401)
        
        token = auth_header[7:]
        if token not in tokens:
            abort(401)
        
        kwargs["token"] = token
        return f(*args, **kwargs)
    return decorated_function

def rate_limit(max_requests=10, window=60):
    """请求限流装饰器"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            client_ip = request.remote_addr
            current_time = time.time()
            
            if client_ip not in request_counts:
                request_counts[client_ip] = []
            
            # 清理过期记录
            request_counts[client_ip] = [
                t for t in request_counts[client_ip]
                if current_time - t < window
            ]
            
            if len(request_counts[client_ip]) >= max_requests:
                abort(429)
            
            request_counts[client_ip].append(current_time)
            return f(*args, **kwargs)
        return decorated_function
    return decorator

@app.route("/api/login", methods=["POST"])
def login():
    """登录获取令牌"""
    data = request.get_json()
    if not data or "username" not in data or "password" not in data:
        abort(400)
    
    user = next((u for u in users if u["name"] == data["username"] and u["password_hash"] == data["password"]), None)
    if not user:
        abort(401)
    
    token = generate_token(user["name"], data["password"])
    return jsonify({"token": token, "expires_in": 3600})

@app.route("/api/users", methods=["GET"])
@require_auth
@rate_limit(max_requests=10, window=60)
def get_users(token):
    """获取所有用户（需要认证）"""
    safe_users = [{k: v for k, v in u.items() if k != "password_hash"} for u in users]
    return jsonify({"users": safe_users, "count": len(safe_users)})

@app.route("/api/data", methods=["GET"])
@require_auth
@rate_limit(max_requests=5, window=60)
def get_data(token):
    """获取敏感数据（更严格的限流）"""
    return jsonify({"data": "这是敏感数据", "token_info": tokens[token]})

if __name__ == "__main__":
    app.run(debug=True, port=5000)

测试这个带认证的API：

import requests

BASE_URL = "http://localhost:5000/api"

# 登录获取令牌
login_response = requests.post(f"{BASE_URL}/login", json={
    "username": "张三",
    "password": "abc123"
})
token = login_response.json()["token"]
print(f"获取令牌: {token}")

# 使用令牌访问受保护的资源
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(f"{BASE_URL}/users", headers=headers)
print("用户列表:", response.json())

# 不使用令牌访问（会失败）
response = requests.get(f"{BASE_URL}/users")
print(f"无令牌访问: {response.status_code}")

# 频繁请求触发限流
for i in range(12):
    response = requests.get(f"{BASE_URL}/data", headers=headers)
    if response.status_code == 429:
        print(f"第{i+1}次请求被限流")
        break
    print(f"第{i+1}次请求: {response.status_code}")

实战：写一个完整的爬虫+API示例

光说不练假把式，咱们来一个综合实战：写一个能抓取 GitHub 仓库信息并对外提供API的服务。

import requests
from flask import Flask, jsonify, request
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import threading

app = Flask(__name__)

# 缓存存储
cache = {}
cache_lock = threading.Lock()
CACHE_TTL = 300  # 缓存5分钟

def fetch_repo_info(username, repo_name):
    """抓取单个仓库信息"""
    url = f"https://api.github.com/repos/{username}/{repo_name}"
    
    headers = {
        "Accept": "application/vnd.github.v3+json",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    response = requests.get(url, headers=headers, timeout=15)
    
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 403:
        # GitHub API限流，等一会再试
        time.sleep(2)
        response = requests.get(url, headers=headers, timeout=15)
        if response.status_code == 200:
            return response.json()
    return None

def get_cached_or_fetch(username, repo_name):
    """获取缓存或抓取"""
    cache_key = f"{username}/{repo_name}"
    
    with cache_lock:
        if cache_key in cache:
            cached_data, cached_time = cache[cache_key]
            if time.time() - cached_time < CACHE_TTL:
                return cached_data
    
    result = fetch_repo_info(username, repo_name)
    
    if result:
        with cache_lock:
            cache[cache_key] = (result, time.time())
    
    return result

@app.route("/api/repos", methods=["GET"])
def search_repos():
    """搜索仓库"""
    query = request.args.get("q", "")
    sort = request.args.get("sort", "stars")
    per_page = request.args.get("per_page", 10)
    
    if not query:
        return jsonify({"error": "请提供搜索关键词"}), 400
    
    url = "https://api.github.com/search/repositories"
    params = {
        "q": query,
        "sort": sort,
        "per_page": int(per_page)
    }
    
    headers = {
        "Accept": "application/vnd.github.v3+json",
        "User-Agent": "RepoSearchBot/1.0"
    }
    
    response = requests.get(url, params=params, headers=headers, timeout=20)
    
    if response.status_code == 200:
        return jsonify(response.json())
    elif response.status_code == 403:
        return jsonify({"error": "GitHub API限流，请稍后重试"}), 429
    else:
        return jsonify({"error": f"请求失败: {response.status_code}"}), response.status_code

@app.route("/api/repos/<username>/<repo_name>", methods=["GET"])
def get_repo(username, repo_name):
    """获取指定仓库信息（带缓存）"""
    result = get_cached_or_fetch(username, repo_name)
    
    if result:
        return jsonify(result)
    else:
        return jsonify({"error": "仓库不存在或获取失败"}), 404

@app.route("/api/repos/batch", methods=["POST"])
def get_repos_batch():
    """批量获取多个仓库信息（并发）"""
    data = request.get_json()
    if not data or "repos" not in data:
        return jsonify({"error": "请提供repos数组"}), 400
    
    repos = data["repos"]  # [{"username": "xxx", "repo_name": "yyy"}, ...]
    
    results = []
    
    # 并发抓取
    with ThreadPoolExecutor(max_workers=10) as executor:
        future_to_repo = {
            executor.submit(get_cached_or_fetch, r["username"], r["repo_name"]): r
            for r in repos
        }
        
        for future in as_completed(future_to_repo):
            repo = future_to_repo[future]
            try:
                result = future.result()
                if result:
                    results.append({
                        "repo": f"{repo['username']}/{repo['repo_name']}",
                        "data": result
                    })
                else:
                    results.append({
                        "repo": f"{repo['username']}/{repo['repo_name']}",
                        "error": "获取失败"
                    })
            except Exception as e:
                results.append({
                    "repo": f"{repo['username']}/{repo['repo_name']}",
                    "error": str(e)
                })
    
    return jsonify({"results": results, "total": len(results)})

@app.route("/api/repos/top", methods=["GET"])
def get_top_repos():
    """获取热门仓库"""
    languages = request.args.get("language", "Python")
    since = request.args.get("since", "daily")
    limit = request.args.get("limit", 10)
    
    url = "https://api.github.com/search/repositories"
    params = {
        "q": f"language:{languages}",
        "sort": "stars",
        "order": "desc",
        "per_page": int(limit)
    }
    
    headers = {
        "Accept": "application/vnd.github.v3+json",
        "User-Agent": "RepoSearchBot/1.0"
    }
    
    response = requests.get(url, params=params, headers=headers, timeout=20)
    
    if response.status_code == 200:
        items = response.json().get("items", [])
        simplified = []
        for item in items:
            simplified.append({
                "name": item["full_name"],
                "description": item["description"],
                "stars": item["stargazers_count"],
                "forks": item["forks_count"],
                "url": item["html_url"],
                "language": item["language"],
                "updated_at": item["updated_at"]
            })
        return jsonify({"repos": simplified, "total": len(simplified)})
    else:
        return jsonify({"error": "获取失败"}), response.status_code

if __name__ == "__main__":
    print("🚀 服务启动在 http://localhost:5000")
    print("📚 API端点:")
    print("  GET /api/repos?q=关键词 - 搜索仓库")
    print("  GET /api/repos/<user>/<repo> - 获取仓库详情")
    print("  POST /api/repos/batch - 批量获取仓库")
    print("  GET /api/repos/top?language=Python - 获取热门仓库")
    app.run(debug=True, port=5000)

这个例子综合了前面讲的所有知识点：

  • 带Headers的请求
  • 超时处理
  • 并发请求
  • 缓存机制
  • 错误处理
  • RESTful API设计

测试这个服务：

import requests
import time

BASE_URL = "http://localhost:5000/api"

# 搜索仓库
print("=== 搜索仓库 ===")
response = requests.get(f"{BASE_URL}/repos", params={"q": "machine learning", "sort": "stars"})
print(response.json())

# 获取单个仓库
print("\n=== 获取仓库详情 ===")
response = requests.get(f"{BASE_URL}/repos/requests/requests")
print(response.json())

# 批量获取
print("\n=== 批量获取 ===")
repos_list = [
    {"username": "requests", "repo_name": "requests"},
    {"username": "python", "repo_name": "cpython"},
    {"username": "torvalds", "repo_name": "linux"},
]
response = requests.post(f"{BASE_URL}/repos/batch", json={"repos": repos_list})
print(response.json())

# 获取热门Python仓库
print("\n=== 热门仓库 ===")
response = requests.get(f"{BASE_URL}/repos/top", params={"language": "Python", "limit": 5})
print(response.json())

错误处理的黄金法则

最后，让我分享一下我踩过的坑总结出来的错误处理经验：

1. 永远不要忽略异常

# 错误的写法
response = requests.get(url)
print(response.text)

# 正确的写法
try:
    response = requests.get(url, timeout=10)
    response.raise_for_status()  # 检查HTTP错误
    print(response.text)
except requests.exceptions.HTTPError as e:
    print(f"HTTP错误: {e}")
except requests.exceptions.Timeout:
    print("请求超时")
except requests.exceptions.ConnectionError:
    print("网络连接失败")
except requests.exceptions.RequestException as e:
    print(f"未知错误: {e}")

2. 使用重试而不是直接失败

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=requests.adapters.HTTPAdapter(max_retries=3)
)
def robust_get(url):
    return requests.get(url, timeout=30)

tenacity库让你的重试逻辑更灵活，可以指定重试次数、等待时间等。

3. 日志记录

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def fetch_with_logging(url):
    logger.info(f"开始请求: {url}")
    try:
        response = requests.get(url, timeout=10)
        logger.info(f"请求成功: {url}, 状态码: {response.status_code}")
        return response
    except requests.exceptions.Timeout:
        logger.warning(f"请求超时: {url}")
        raise
    except requests.exceptions.ConnectionError:
        logger.error(f"连接失败: {url}")
        raise

4. 区分可重试和不可重试的错误

def should_retry(status_code):
    """判断是否需要重试"""
    # 5xx服务器错误和429限流可以重试
    return status_code >= 500 or status_code == 429

def fetch_smart(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=10)
            
            if response.status_code == 200:
                return response
            elif should_retry(response.status_code):
                wait_time = 2 ** attempt  # 指数退避
                print(f"第{attempt+1}次尝试失败({response.status_code})，{wait_time}秒后重试...")
                time.sleep(wait_time)
                continue
            else:
                # 4xx客户端错误不需要重试
                return response
        except requests.exceptions.Timeout:
            if attempt < max_retries - 1:
                print(f"超时，{2 ** attempt}秒后重试...")
                time.sleep(2 ** attempt)
            else:
                raise
    return None

写在最后

好了，今天的干货就到这里。从最基本的GET请求到复杂的并发、缓存、认证、限流，我把HTTP编程里最常用的东西都讲到了。记住，网络编程没有捷径，就是多写多练，遇到错误不要怕，一个个解决。

你现在的目标应该是：

  1. 把这个例子跑起来，亲自测试每一个功能
  2. 尝试修改代码，加入新的功能
  3. 找一个真实的API项目练手

有任何问题，随时来问我。祝你玩得开心！

-- 展开阅读全文 --