HTTP协议网络编程实例大全从Requests基础到服务器搭建手把手教你实现客户端与服务端通信爬虫接口开发Web服务构建完整代码详解
好家伙，说到HTTP编程，很多人第一反应就是”又要背那些死板的语法了”。别急，今天咱们不整那些虚头巴脑的教科书式讲解，我直接带着你从最基础的请求发起到自己搭建服务器，一步步把整个HTTP编程的体系摸透。我干了这么多年网络编程，见过太多人卡在第一步就放弃了，其实HTTP这东西没那么可怕，你只需要搞清楚它在干嘛就行。
咱们先聊聊HTTP到底是干嘛的
HTTP说白了就是一种约定。你和我约定好，我发一个请求过去，你给我一个响应回来。就这么简单。你浏览器打开一个网页，背后就是成百上千个HTTP请求在跑。
import requests
# 最简单的一个GET请求，就这么一行代码
response = requests.get("https://httpbin.org/get")
# 看看返回了啥
print(f"状态码: {response.status_code}")
print(f"响应内容: {response.text[:200]}...")
运行这段代码，你会看到类似这样的输出：
状态码: 200
响应内容: {"args": {}, "headers": {"Accept": "*/*", "Accept-Encoding": "gzip, deflate",
"Host": "httpbin.org", "User-Agent": "python-requests/2.28.1"}, ...
看到了吧，200就是告诉你对了，请求成功。你要是看到404，那就是”哎呀，页面没找到”。500的话，就是服务器内部出问题了。这些状态码就是HTTP世界里的一套暗号，后面咱们会一个个拆解。
Requests库的十种姿势，我带你一个个试
2.1 GET请求——最最常见的请求方式
GET就是”给我数据”，它是幂等的，也就是说你发一百次，结果都一样。
import requests
import json
# 基本GET请求
url = "https://httpbin.org/get"
response = requests.get(url)
print("=== 基本GET请求 ===")
print(f"状态码: {response.status_code}")
print(f"响应头中的Content-Type: {response.headers.get('Content-Type')}")
# 带参数的GET请求，比如搜索
search_url = "https://httpbin.org/get"
params = {
"name": "张三",
"age": 25,
"keyword": "Python教程"
}
response = requests.get(search_url, params=params)
print("\n=== 带参数的GET请求 ===")
print(f"请求URL: {response.request.url}")
print(f"响应JSON: {json.dumps(response.json(), indent=2, ensure_ascii=False)}")
运行结果里你会看到，参数被自动拼接到了URL后面：
请求URL: https://httpbin.org/get?name=%E5%BC%A0%E4%B8%89&age=25&keyword=Python%E6%95%99%E7%A8%8B
httpbin.org 这个网站特别好，它是专门用来测试HTTP请求的，你发的任何东西它都会原样返回给你，简直是学习HTTP的福音。
2.2 POST请求——提交数据给服务器
POST就是”我给你数据，你帮我处理”，和GET最大的区别是数据放在请求体里，不在URL里，适合提交敏感信息或者大数据。
import requests
import json
# 基本POST请求
url = "https://httpbin.org/post"
# 方式一：通过data参数提交表单数据
form_data = {
"username": "xiaoming",
"password": "123456",
"remember": "true"
}
response = requests.post(url, data=form_data)
print("=== POST表单请求 ===")
print(f"状态码: {response.status_code}")
print(f"提交的表单数据: {json.dumps(response.json().get('form'), indent=2)}")
# 方式二：通过json参数提交JSON数据（现在更常见的做法）
json_data = {
"title": "HTTP编程完全指南",
"author": "老张",
"tags": ["Python", "网络编程", "HTTP"],
"price": 99.9
}
response = requests.post(url, json=json_data)
print("\n=== POST JSON请求 ===")
print(f"提交的JSON数据: {json.dumps(response.json().get('json'), indent=2)}")
print(f"Content-Type: {response.request.headers.get('Content-Type')}")
你会发现用json=参数和用data=参数，服务器收到的数据格式是完全不同的。json=会自动把数据序列化成JSON字符串，并且设置Content-Type: application/json，这是现在API交互的主流方式。
2.3 其他请求方法——PUT、DELETE、PATCH
import requests
base_url = "https://httpbin.org"
# PUT：全量更新资源
print("=== PUT请求 ===")
response = requests.put(f"{base_url}/put", json={"id": 1, "name": "李四", "age": 30})
print(f"状态码: {response.status_code}")
print(f"收到的JSON: {response.json().get('json')}")
# DELETE：删除资源
print("\n=== DELETE请求 ===")
response = requests.delete(f"{base_url}/delete")
print(f"状态码: {response.status_code}")
# PATCH：局部更新资源（只更新部分字段）
print("\n=== PATCH请求 ===")
response = requests.patch(f"{base_url}/patch", json={"age": 31})
print(f"状态码: {response.status_code}")
print(f"收到的JSON: {response.json().get('json')}")
# HEAD：只获取响应头，不获取响应体（常用于检查资源是否存在）
print("\n=== HEAD请求 ===")
response = requests.head(f"{base_url}/get")
print(f"状态码: {response.status_code}")
print(f"响应头: {dict(response.headers)}")
# OPTIONS：查询服务器支持哪些请求方法
print("\n=== OPTIONS请求 ===")
response = requests.options(f"{base_url}/get")
print(f"允许的方法: {response.headers.get('Allow')}")
2.4 带Headers的请求——伪造浏览器身份
很多网站会检查你的User-Agent，如果你用Python默认的标识去访问，很可能被拒之门外。
import requests
url = "https://httpbin.org/headers"
# 模拟浏览器请求，带上完整的Headers
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,image/apng,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
"Cache-Control": "max-age=0",
"Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." # 认证token
}
response = requests.get(url, headers=headers)
print("=== 带Headers的请求 ===")
print(f"服务器收到的Headers: {response.json().get('headers')}")
2.5 处理Cookie和Session——保持登录状态
import requests
# 方式一：手动设置Cookie
url = "https://httpbin.org/cookies"
cookies = {"session_id": "abc123xyz", "user": "testuser"}
response = requests.get(url, cookies=cookies)
print("=== 手动Cookie ===")
print(f"返回的Cookie: {response.json()}")
# 方式二：Session保持会话（自动管理Cookie）
session = requests.Session()
# 第一次请求，设置Cookie
session.cookies.set("theme", "dark", domain="httpbin.org")
response = session.get("https://httpbin.org/cookies")
print(f"\nSession Cookie: {response.json()}")
# 第二次请求，Session自动携带Cookie
response = session.get("https://httpbin.org/cookies")
print(f"再次请求: {response.json()}")
# 模拟登录流程
login_url = "https://httpbin.org/post"
login_data = {"username": "admin", "password": "secret123"}
response = session.post(login_url, data=login_data)
print(f"\n登录响应状态: {response.status_code}")
# 登录后访问需要认证的页面
profile_url = "https://httpbin.org/user-info"
response = session.get(profile_url)
print(f"用户信息页面状态: {response.status_code}")
2.6 超时设置——别让你的程序卡死
import requests
from requests.exceptions import Timeout, ConnectionError, HTTPError
url = "https://httpbin.org/delay/5"
try:
# timeout可以是一个值（连接+读取的总超时），也可以是元组（连接超时, 读取超时）
response = requests.get(url, timeout=(3, 5))
print(f"请求成功，状态码: {response.status_code}")
except Timeout:
print("⏰ 请求超时了！服务器5秒内没响应")
except ConnectionError:
print("🔌 网络连接错误，检查一下网络")
except HTTPError as e:
print(f"❌ HTTP错误: {e}")
except Exception as e:
print(f"💥 其他错误: {e}")
# 实际项目中，建议这样设置
timeout_config = {
"connect_timeout": 10, # 连接超时10秒
"read_timeout": 30, # 读取超时30秒
"total_timeout": 60 # 总超时60秒
}
print(f"\n推荐的超时配置: 连接{timeout_config['connect_timeout']}秒, 读取{timeout_config['read_timeout']}秒")
2.7 文件上传——上传图片、文档
import requests
url = "https://httpbin.org/post"
# 上传单个文件
with open("test.txt", "w") as f:
f.write("这是要上传的文件内容\nHello HTTP World!")
with open("test.txt", "rb") as f:
files = {"file": ("test.txt", f, "text/plain")}
response = requests.post(url, files=files)
print("=== 文件上传 ===")
print(f"上传状态: {response.status_code}")
print(f"上传的文件名: {response.json().get('files', {}).get('file')}")
# 上传多个文件
with open("image.png", "wb") as f: # 创建一个假图片文件
f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
multi_files = {
"document": ("report.pdf", open("test.txt", "rb"), "application/pdf"),
"image": ("avatar.png", open("image.png", "rb"), "image/png")
}
response = requests.post(url, files=multi_files)
print(f"\n多文件上传状态: {response.status_code}")
# 带元数据的文件上传
files = {"file": ("data.csv", b"name,age\n张三,25\n李四,30", "text/csv")}
response = requests.post(url, files=files, data={"description": "用户数据表"})
print(f"带元数据上传: {response.json().get('form')}")
2.8 下载文件——大文件断点续传
import requests
import os
url = "https://httpbin.org/stream/100" # 模拟大文件流
# 基本下载
response = requests.get(url, stream=True)
if response.status_code == 200:
with open("downloaded_data.txt", "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print("✅ 文件下载完成")
# 带进度条的下载
def download_with_progress(url, filepath, chunk_size=8192):
response = requests.get(url, stream=True)
total_size = int(response.headers.get("content-length", 0))
with open(filepath, "wb") as f:
downloaded = 0
for chunk in response.iter_content(chunk_size):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_size > 0:
percent = downloaded * 100 / total_size
bar = "█" * int(percent / 2) + "░" * (50 - int(percent / 2))
print(f"\r📥 下载进度: [{bar}] {percent:.1f}% ({downloaded}/{total_size} bytes)", end="")
print("\n✅ 下载完成！")
# download_with_progress(url, "output.bin") # 取消注释即可运行
2.9 SSL证书验证——安全连接的处理
import requests
from requests.exceptions import SSLError
# 默认情况下，requests会验证SSL证书
try:
response = requests.get("https://httpbin.org/get", verify=True)
print(f"✅ SSL验证通过，状态码: {response.status_code}")
except SSLError as e:
print(f"❌ SSL验证失败: {e}")
# 跳过SSL验证（开发环境可以，生产环境不建议）
try:
response = requests.get("https://self-signed.badssl.com/", verify=False)
print(f"⚠️ 跳过SSL验证，状态码: {response.status_code}")
except Exception as e:
print(f"即使跳过验证也出错了: {e}")
# 使用自定义CA证书
# response = requests.get("https://example.com", verify="/path/to/ca-bundle.crt")
# 关闭不安全的警告（使用verify=False时会触发）
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
2.10 代理设置——绕过网络限制
import requests
# 使用HTTP代理
proxies = {
"http": "http://127.0.0.1:7890",
"https": "https://127.0.0.1:7890"
}
try:
response = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=10)
print(f"通过代理获取的IP: {response.json()}")
except Exception as e:
print(f"代理请求失败: {e}")
# 使用 socks 代理
# proxies = {
# "http": "socks5://127.0.0.1:1080",
# "https": "socks5://127.0.0.1:1080"
# }
# 需要认证的代理
auth_proxies = {
"http": "http://username:password@proxy.example.com:8080",
"https": "http://username:password@proxy.example.com:8080"
}
# 环境变量方式设置代理（系统级别）
# import os
# os.environ["HTTP_PROXY"] = "http://proxy.example.com:8080"
# os.environ["HTTPS_PROXY"] = "https://proxy.example.com:8080"
爬虫实战——从零基础到真正能跑
3.1 第一个爬虫——抓取网页标题
import requests
from bs4 import BeautifulSoup
import re
def scrape_title(url):
"""抓取网页标题"""
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"
}
try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status() # 如果状态码不是200，抛出异常
response.encoding = response.apparent_encoding # 自动检测编码
soup = BeautifulSoup(response.text, "html.parser")
title = soup.title.string if soup.title else "无标题"
return {
"url": url,
"title": title,
"status_code": response.status_code,
"content_length": len(response.text)
}
except Exception as e:
return {"url": url, "error": str(e)}
# 测试
result = scrape_title("https://example.com")
print(f"网页标题: {result['title']}")
print(f"状态码: {result['status_code']}")
print(f"内容长度: {result['content_length']} 字节")
3.2 批量爬虫——抓取多个页面的数据
import requests
from bs4 import BeautifulSoup
import time
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
class SimpleCrawler:
def __init__(self, delay=1, max_retries=3):
self.delay = delay # 请求间隔，避免被封
self.max_retries = max_retries
self.session = requests.Session()
self.session.headers.update({
"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"
})
def fetch_page(self, url):
"""带重试的页面抓取"""
for attempt in range(self.max_retries):
try:
response = self.session.get(url, timeout=15)
response.raise_for_status()
response.encoding = response.apparent_encoding
time.sleep(self.delay) # 礼貌爬虫，别太猛
return response
except requests.exceptions.RequestException as e:
if attempt == self.max_retries - 1:
print(f"❌ {url} 重试{self.max_retries}次后仍失败: {e}")
return None
wait_time = 2 ** attempt # 指数退避
print(f"⚠️ {url} 第{attempt+1}次失败，{wait_time}秒后重试...")
time.sleep(wait_time)
return None
def extract_articles(self, html, source_url):
"""从HTML中提取文章信息"""
soup = BeautifulSoup(html, "html.parser")
articles = []
# 这里以知乎热榜为例，实际抓取需要根据目标网站调整选择器
# 通用模式：查找文章列表
for item in soup.select(".listitem, .feed-item, .article-item, [class*='item']"):
link_tag = item.select_one("a[href]")
if link_tag and link_tag.get("href"):
title = link_tag.get_text(strip=True)
href = link_tag["href"]
if not href.startswith("http"):
href = source_url.rstrip("/") + "/" + href.lstrip("/")
articles.append({
"title": title,
"url": href,
"source": source_url
})
return articles
def crawl(self, urls):
"""批量爬取"""
all_articles = []
total = len(urls)
print(f"🕷️ 开始爬取 {total} 个页面...")
for idx, url in enumerate(urls, 1):
print(f"📄 [{idx}/{total}] 正在抓取: {url[:50]}...")
response = self.fetch_page(url)
if response:
articles = self.extract_articles(response.text, url)
all_articles.extend(articles)
print(f" ✅ 提取到 {len(articles)} 条数据")
return all_articles
def save_results(self, articles, filename="results.json"):
"""保存结果到JSON文件"""
with open(filename, "w", encoding="utf-8") as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
print(f"💾 结果已保存到 {filename}")
# 使用示例
if __name__ == "__main__":
urls = [
"https://example.com",
"https://httpbin.org/html"
]
crawler = SimpleCrawler(delay=1)
results = crawler.crawl(urls)
crawler.save_results(results)
print(f"\n🎉 爬取完成！共获得 {len(results)} 条数据")
3.3 动态网页爬虫——处理JavaScript渲染的内容
很多现代网站的数据是JavaScript动态加载的，普通requests拿不到。这时候有两个选择：Selenium 和 直接调接口。
# 方案一：直接找API接口（推荐，更高效）
import requests
def crawl_api_directly():
"""很多网站的前端数据都来自隐藏的API，直接调API效率更高"""
# 假设我们想抓某个音乐平台的歌曲列表
api_url = "https://httpbin.org/post"
# 模拟前端发送的API请求
payload = {
"pageSize": 20,
"pageNo": 1,
"category": "pop"
}
headers = {
"Content-Type": "application/json",
"X-Requested-With": "XMLHttpRequest",
"Referer": "https://example.com/music"
}
response = requests.post(api_url, json=payload, headers=headers)
print(f"API响应: {response.json()}")
# crawl_api_directly()
# 方案二：Selenium处理动态渲染（当API找不到时的备选）
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time
def crawl_dynamic_page():
"""使用Selenium处理JavaScript渲染的页面"""
options = webdriver.ChromeOptions()
options.add_argument("--headless") # 无头模式，不显示浏览器窗口
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--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")
driver = webdriver.Chrome(options=options)
try:
driver.get("https://example.com")
# 等待页面加载完成
WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.TAG_NAME, "body"))
)
# 等待动态内容加载
time.sleep(3)
# 提取数据
title = driver.title
content = driver.find_element(By.TAG_NAME, "body").text
print(f"页面标题: {title}")
print(f"页面内容长度: {len(content)} 字符")
# 滚动加载更多
for _ in range(3):
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
time.sleep(1)
final_content = driver.find_element(By.TAG_NAME, "body").text
print(f"滚动后内容长度: {len(final_content)} 字符")
finally:
driver.quit()
# crawl_dynamic_page() # 需要安装selenium和chromedriver
3.4 反爬虫应对策略——让你的爬虫更稳
import requests
import random
import time
from fake_useragent import UserAgent
class AntiDetectionCrawler:
"""带反检测能力的爬虫"""
def __init__(self):
self.session = requests.Session()
self.ua = UserAgent()
self.delay_range = (1, 3) # 随机延迟
# 轮换的Headers池
self.headers_pool = [
{
"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/*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
},
{
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) "
"Version/17.0 Safari/605.1.15",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9/*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
},
{
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/119.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9/*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.5",
}
]
def get_random_headers(self):
"""随机获取Headers"""
headers = random.choice(self.headers_pool).copy()
# 偶尔添加一些随机字段，更像真实浏览器
if random.random() < 0.3:
headers["DNT"] = "1"
if random.random() < 0.2:
headers["Upgrade-Insecure-Requests"] = "1"
return headers
def get(self, url, **kwargs):
"""带反检测的GET请求"""
kwargs.setdefault("headers", self.get_random_headers())
kwargs.setdefault("timeout", 15)
delay = random.uniform(*self.delay_range)
time.sleep(delay)
try:
response = self.session.get(url, **kwargs)
return response
except requests.exceptions.RequestException as e:
print(f"请求失败 {url}: {e}")
return None
def get_with_retry(self, url, max_retries=3, **kwargs):
"""带重试的请求"""
for attempt in range(max_retries):
response = self.get(url, **kwargs)
if response and response.status_code == 200:
return response
print(f"第{attempt + 1}次请求失败，等待重试...")
time.sleep(random.uniform(2, 5))
return None
# 使用示例
# crawler = AntiDetectionCrawler()
# response = crawler.get("https://httpbin.org/headers")
# print(response.json())
接口开发——用Flask构建自己的HTTP服务
4.1 最小可用的Web服务
from flask import Flask, jsonify, request, abort
app = Flask(__name__)
# 模拟数据库
users = [
{"id": 1, "name": "张三", "email": "zhangsan@example.com", "age": 25},
{"id": 2, "name": "李四", "email": "lisi@example.com", "age": 30},
{"id": 3, "name": "王五", "email": "wangwu@example.com", "age": 28},
]
@app.route("/")
def index():
"""首页"""
return jsonify({
"message": "欢迎使用HTTP编程实战API",
"version": "1.0.0",
"endpoints": {
"GET /users": "获取用户列表",
"GET /users/<id>": "获取单个用户",
"POST /users": "创建用户",
"PUT /users/<id>": "更新用户",
"DELETE /users/<id>": "删除用户"
}
})
@app.route("/users", methods=["GET"])
def get_users():
"""获取所有用户"""
return jsonify({
"code": 200,
"message": "success",
"data": users,
"total": len(users)
})
@app.route("/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 not user:
abort(404)
return jsonify({"code": 200, "data": user})
@app.route("/users", methods=["POST"])
def create_user():
"""创建用户"""
data = request.get_json()
# 参数验证
if not data:
abort(400, description="请求体不能为空")
if "name" not in data or "email" not in data:
abort(400, description="缺少必要字段: name, email")
new_user = {
"id": max(u["id"] for u in users) + 1 if users else 1,
"name": data["name"],
"email": data["email"],
"age": data.get("age", 0)
}
users.append(new_user)
return jsonify({"code": 201, "message": "创建成功", "data": new_user}), 201
@app.route("/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:
abort(404)
data = request.get_json()
if data:
if "name" in data:
user["name"] = data["name"]
if "email" in data:
user["email"] = data["email"]
if "age" in data:
user["age"] = data["age"]
return jsonify({"code": 200, "message": "更新成功", "data": user})
@app.route("/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({"code": 200, "message": "删除成功"})
@app.errorhandler(404)
def not_found(e):
return jsonify({"code": 404, "message": str(e.description)}), 404
@app.errorhandler(500)
def internal_error(e):
return jsonify({"code": 500, "message": "服务器内部错误"}), 500
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5000)
启动后你可以这样测试：
# 获取用户列表
curl http://localhost:5000/users
# 创建用户
curl -X POST http://localhost:5000/users \
-H "Content-Type: application/json" \
-d '{"name": "赵六", "email": "zhaoliu@example.com", "age": 22}'
# 获取单个用户
curl http://localhost:5000/users/1
# 更新用户
curl -X PUT http://localhost:5000/users/1 \
-H "Content-Type: application/json" \
-d '{"name": "张三丰", "age": 100}'
# 删除用户
curl -X DELETE http://localhost:5000/users/1
4.2 带认证和权限的API
from flask import Flask, jsonify, request, abort
import hashlib
import time
import jwt
from functools import wraps
app = Flask(__name__)
app.config["SECRET_KEY"] = "your-secret-key-change-in-production"
# 模拟用户数据库
users_db = {
"admin": {"password": hashlib.sha256(b"admin123").hexdigest(), "role": "admin"},
"user1": {"password": hashlib.sha256(b"password123").hexdigest(), "role": "user"}
}
def generate_token(username, role, expiry_hours=24):
"""生成JWT Token"""
payload = {
"user": username,
"role": role,
"exp": time.time() + expiry_hours * 3600,
"iat": time.time()
}
return jwt.encode(payload, app.config["SECRET_KEY"], algorithm="HS256")
def verify_token(token):
"""验证JWT Token"""
try:
payload = jwt.decode(token, app.config["SECRET_KEY"], algorithms=["HS256"])
return payload
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
def login_required(f):
"""登录验证装饰器"""
@wraps(f)
def decorated_function(*args, **kwargs):
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
abort(401, description="缺少认证信息")
token = auth_header[7:] # 去掉 "Bearer " 前缀
payload = verify_token(token)
if not payload:
abort(401, description="Token无效或已过期")
request.user = payload # 将用户信息存入request
return f(*args, **kwargs)
return decorated_function
def role_required(role):
"""角色权限验证"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if getattr(request, "user", {}).get("role") != role:
abort(403, description="权限不足")
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route("/api/login", methods=["POST"])
def login():
"""用户登录，获取Token"""
data = request.get_json()
if not data or "username" not in data or "password" not in data:
abort(400, description="缺少用户名或密码")
username = data["username"]
password_hash = hashlib.sha256(data["password"].encode()).hexdigest()
user = users_db.get(username)
if not user or user["password"] != password_hash:
abort(401, description="用户名或密码错误")
token = generate_token(username, user["role"])
return jsonify({
"code": 200,
"message": "登录成功",
"data": {
"token": token,
"username": username,
"role": user["role"]
}
})
@app.route("/api/protected")
@login_required
def protected_resource():
"""需要登录才能访问的资源"""
return jsonify({
"code": 200,
"message": "恭喜你，访问成功！",
"data": {
"user": request.user["user"],
"role": request.user["role"]
}
})
@app.route("/api/admin/users", methods=["GET"])
@login_required
@role_required("admin")
def get_all_users():
"""只有管理员能访问的用户列表"""
return jsonify({
"code": 200,
"data": list(users_db.keys())
})
if __name__ == "__main__":
app.run(debug=True, port=5001)
测试认证API：
# 1. 登录获取Token
TOKEN=$(curl -s -X POST http://localhost:5001/api/login \
-H "Content-Type: application/json" \
-d '{"username": "admin", "password": "admin123"}' | \
jq -r '.data.token')
# 2. 用Token访问保护接口
curl http://localhost:5001/api/protected \
-H "Authorization: Bearer $TOKEN"
# 3. 没有Token访问（应该401）
curl http://localhost:5001/api/protected
4.3 RESTful API的最佳实践
from flask import Flask, jsonify, request, abort
from flask_cors import CORS
from marshmallow import Schema, fields, validate, ValidationError
import uuid
app = Flask(__name__)
CORS(app) # 跨域支持
# 请求验证Schema
class UserCreateSchema(Schema):
name = fields.String(required=True, validate=validate.Length(min=1, max=50))
email = fields.Email(required=True)
age = fields.Integer(validate=validate.Range(min=0, max=150))
class UserUpdateSchema(Schema):
name = fields.String(validate=validate.Length(min=1, max=50))
email = fields.Email()
age = fields.Integer(validate=validate.Range(min=0, max=150))
user_create_schema = UserCreateSchema()
user_update_schema = UserUpdateSchema()
# 内存存储
articles = {}
@app.route("/api/v1/articles", methods=["GET"])
def list_articles():
"""
获取文章列表
Query参数: page, page_size, category, keyword
"""
page = request.args.get("page", 1, type=int)
page_size = request.args.get("page_size", 10, type=int)
category = request.args.get("category")
keyword = request.args.get("keyword")
# 过滤
filtered = articles.values()
if category:
filtered = [a for a in filtered if a["category"] == category]
if keyword:
filtered = [a for a in filtered if keyword in a["title"]]
# 分页
total = len(filtered)
start = (page - 1) * page_size
end = start + page_size
page_articles = list(filtered)[start:end]
return jsonify({
"code": 200,
"data": {
"items": page_articles,
"pagination": {
"page": page,
"page_size": page_size,
"total": total,
"total_pages": (total + page_size - 1) // page_size
}
}
})
@app.route("/api/v1/articles/<article_id>", methods=["GET"])
def get_article(article_id):
"""获取单篇文章"""
article = articles.get(article_id)
if not article:
abort(404, description=f"文章 {article_id} 不存在")
return jsonify({"code": 200, "data": article})
@app.route("/api/v1/articles", methods=["POST"])
def create_article():
"""创建文章"""
data = request.get_json()
# 参数验证
try:
validated_data = user_create_schema.load(data) # 注意：实际应该用ArticleCreateSchema
except ValidationError as e:
abort(400, description=str(e.messages))
article_id = str(uuid.uuid4())
article = {
"id": article_id,
"title": validated_data["name"],
"content": data.get("content", ""),
"category": data.get("category", "general"),
"author": "anonymous",
"created_at": time_iso(),
"updated_at": time_iso()
}
articles[article_id] = article
return jsonify({"code": 201, "data": article}), 201
@app.route("/api/v1/articles/<article_id>", methods=["PUT"])
def update_article(article_id):
"""更新文章"""
if article_id not in articles:
abort(404)
data = request.get_json()
try:
validated_data = user_update_schema.load(data)
except ValidationError as e:
abort(400, description=str(e.messages))
article = articles[article_id]
if "name" in validated_data:
article["title"] = validated_data["name"]
if "email" in validated_data:
article["author"] = validated_data["email"]
article["updated_at"] = time_iso()
return jsonify({"code": 200, "data": article})
@app.route("/api/v1/articles/<article_id>", methods=["DELETE"])
def delete_article(article_id):
"""删除文章"""
if article_id not in articles:
abort(404)
del articles[article_id]
return jsonify({"code": 200, "message": "删除成功"})
def time_iso():
from datetime import datetime
return datetime.utcnow().isoformat() + "Z"
@app.errorhandler(400)
def bad_request(e):
return jsonify({"code": 400, "message": e.description}), 400
@app.errorhandler(404)
def not_found(e):
return jsonify({"code": 404, "message": e.description}), 404
@app.errorhandler(500)
def internal_error(e):
return jsonify({"code": 500, "message": "服务器内部错误"}), 500
WebSocket实时通信——从HTTP到双向通道
HTTP是请求-响应模式，服务器不能主动发消息。但有时候你需要服务器主动推数据，比如聊天室、股票行情、实时通知。这时候WebSocket就派上用场了。
5.1 用Flask-SocketIO搭建实时聊天室
from flask import Flask, render_template_string
from flask_socketio import SocketIO, emit, join_room, leave_room
import datetime
app = Flask(__name__)
app.config["SECRET_KEY"] = "secret!"
socketio = SocketIO(app, cors_allowed_origins="*")
# 存储在线用户
users = {}
# 简单的聊天页面HTML
CHAT_PAGE = """
<!DOCTYPE html>
<html>
<head>
<title>实时聊天室</title>
<style>
body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
#messages { border: 1px solid #ccc; height: 400px; overflow-y: scroll; padding: 10px; margin-bottom: 10px; }
.message { margin: 5px 0; padding: 5px; border-radius: 5px; }
.message.system { background: #f0f0f0; color: #666; }
.message .time { color: #999; font-size: 12px; }
input, button { padding: 10px; margin: 5px; }
#status { color: green; }
</style>
</head>
<body>
<h2>🏠 实时聊天室</h2>
<div id="status">状态: <span id="conn-status">未连接</span></div>
<div id="messages"></div>
<input type="text" id="username" placeholder="输入你的名字" style="width: 150px;">
<button onclick="joinChat()">加入聊天</button>
<br>
<input type="text" id="message" placeholder="输入消息..." style="width: 400px;" disabled>
<button onclick="sendMessage()" id="send-btn" disabled>发送</button>
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.7.2/socket.io.js"></script>
<script>
const socket = io();
let username = "";
socket.on('connect', () => {
document.getElementById('conn-status').textContent = '已连接';
document.getElementById('conn-status').style.color = 'green';
});
socket.on('disconnect', () => {
document.getElementById('conn-status').textContent = '已断开';
document.getElementById('conn-status').style.color = 'red';
});
socket.on('system_message', (data) => {
addMessage(data.message, 'system');
});
socket.on('chat_message', (data) => {
addMessage(`<strong>${data.username}:</strong> ${data.message}`, 'user');
});
function joinChat() {
username = document.getElementById('username').value.trim();
if (username) {
socket.emit('join', { username: username });
document.getElementById('message').disabled = false;
document.getElementById('send-btn').disabled = false;
document.getElementById('message').focus();
}
}
function sendMessage() {
const msg = document.getElementById('message').value.trim();
if (msg && username) {
socket.emit('chat_message', { message: msg });
document.getElementById('message').value = "";
}
}
function addMessage(html, type) {
const messages = document.getElementById('messages');
const time = new Date().toLocaleTimeString();
messages.innerHTML += `<div class="message ${type}">${html} <span class="time">[${time}]</span></div>`;
messages.scrollTop = messages.scrollHeight;
}
document.getElementById('message').addEventListener('keypress', (e) => {
if (e.key === 'Enter') sendMessage();
});
</script>
</body>
</html>
"""
@socketio.on("join")
def handle_join(data):
username = data.get("username")
if username:
join_room(username)
users[username] = {"socket_id": request.sid, "joined_at": datetime.datetime.now()}
emit("system_message", {"message": f"🎉 {username} 加入了聊天室"}, broadcast=True, include_self=False)
emit("system_message", {"message": f"当前在线人数: {len(users)}"})
@socketio.on("chat_message")
def handle_chat_message(data):
username = request.sid # 实际应该从session获取
message = data.get("message", "")
if message:
# 找到发送者的用户名
sender_name = None
for name, info in users.items():
if info["socket_id"] == request.sid:
sender_name = name
break
if sender_name:
emit("chat_message", {"username": sender_name, "message": message}, broadcast=True, include_self=False)
@socketio.on("leave")
def handle_leave():
for name, info in list(users.items()):
if info["socket_id"] == request.sid:
del users[name]
emit("system_message", {"message": f"👋 {name} 离开了聊天室"}, broadcast=True, include_self=False)
break
@app.route("/")
def index():
return render_template_string(CHAT_PAGE)
if __name__ == "__main__":
socketio.run(app, debug=True, host="0.0.0.0", port=5002)
5.2 用WebSocket实现实时数据推送
from flask import Flask, render_template_string
from flask_socketio import SocketIO, emit
import threading
import time
import random
app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")
# 模拟股票数据
stocks = {
"AAPL": 150.00,
"GOOGL": 2800.00,
"MSFT": 300.00,
"TSLA": 200.00
}
@app.route("/")
def index():
return """
<!DOCTYPE html>
<html>
<head><title>实时股票行情</title></head>
<body>
<h2>📈 实时股票行情</h2>
<table border="1" cellpadding="10">
<tr><th>股票代码</th><th>价格</th><th>涨跌</th></tr>
<tr><td id="AAPL">150.00</td><td id="AAPL-change">-</td></tr>
<tr><td id="GOOGL">2800.00</td><td id="GOOGL-change">-</td></tr>
<tr><td id="MSFT">300.00</td><td id="MSFT-change">-</td></tr>
<tr><td id="TSLA">200.00</td><td id="TSLA-change">-</td></tr>
</table>
<p id="last-update">最后更新: --</p>
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.7.2/socket.io.js"></script>
<script>
const socket = io();
socket.on('stock_update', (data) => {
document.getElementById(data.symbol).textContent = data.price.toFixed(2);
const changeEl = document.getElementById(data.symbol + '-change');
changeEl.textContent = (data.change >= 0 ? '+' : '') + data.change.toFixed(2);
changeEl.style.color = data.change >= 0 ? 'red' : 'green';
document.getElementById('last-update').textContent =
'最后更新: ' + new Date().toLocaleTimeString();
});
</script>
</body>
</html>
"""
def stock_simulator():
"""后台线程模拟股票价格波动"""
while True:
for symbol in stocks:
old_price = stocks[symbol]
change = random.uniform(-2, 2)
new_price = old_price + change
stocks[symbol] = new_price
socketio.emit("stock_update", {
"symbol": symbol,
"price": new_price,
"change": change
})
time.sleep(2)
# 启动模拟线程
threading.Thread(target=stock_simulator, daemon=True).start()
if __name__ == "__main__":
socketio.run(app, host="0.0.0.0", port=5003)
gRPC——高性能RPC框架实战
当你的服务之间需要高效通信时，HTTP+JSON可能不够快。gRPC基于HTTP/2和Protocol Buffers，性能更好。
6.1 定义Proto文件
// calculator.proto
syntax = "proto3";
package calculator;
service Calculator {
// 简单加法
rpc Add (AddRequest) returns (AddResponse);
// 简单减法
rpc Subtract (SubtractRequest) returns (SubtractResponse);
// 流式加法（客户端流）
rpc StreamingAdd (stream Number) returns (Number);
// 双向流
rpc StreamingCalc (stream CalcRequest) returns (stream CalcResponse);
}
message AddRequest {
double a = 1;
double b = 2;
}
message AddResponse {
double result = 1;
}
message SubtractRequest {
double a = 1;
double b = 2;
}
message SubtractResponse {
double result = 1;
}
message Number {
double value = 1;
}
message CalcRequest {
double a = 1;
string op = 2; // "+", "-", "*", "/"
double b = 3;
}
message CalcResponse {
double result = 1;
string op = 2;
}
6.2 服务端实现
import grpc
from concurrent import futures
import calculator_pb2
import calculator_pb2_grpc
class CalculatorService(calculator_pb2_grpc.CalculatorServicer):
def Add(self, request, context):
result = request.a + request.b
return calculator_pb2.AddResponse(result=result)
def Subtract(self, request, context):
result = request.a - request.b
return calculator_pb2.SubtractResponse(result=result)
def StreamingAdd(self, request_iterator, context):
"""客户端流式：客户端发送多个数字，服务端返回总和"""
total = 0.0
for number in request_iterator:
total += number.value
return calculator_pb2.Number(value=total)
def StreamingCalc(self, request_iterator, context):
"""双向流：客户端发送计算请求，服务端实时返回结果"""
for request in request_iterator:
result = 0.0
if request.op == "+":
result = request.a + request.b
elif request.op == "-":
result = request.a - request.b
elif request.op == "*":
result = request.a * request.b
elif request.op == "/":
if request.b == 0:
context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
context.set_details("除数不能为零")
return
result = request.a / request.b
else:
context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
context.set_details(f"不支持的操作: {request.op}")
return
yield calculator_pb2.CalcResponse(result=result, op=request.op)
def serve():
interpreter = grpc.insecure_server(
futures.ThreadPoolExecutor(max_workers=10),
["0.0.0.0:50051"]
)
calculator_pb2_grpc.add_CalculatorServicer_to_server(
CalculatorService(), interpreter
)
print("🚀 gRPC服务器启动在 0.0.0.0:50051")
interpreter.wait_for_termination()
if __name__ == "__main__":
serve()
6.3 客户端实现
import grpc
import calculator_pb2
import calculator_pb2_grpc
def unary_calls():
"""普通RPC调用"""
channel = grpc.insecure_channel("localhost:50051")
stub = calculator_pb2_grpc.CalculatorStub(channel)
# 加法
response = stub.Add(calculator_pb2.AddRequest(a=10.5, b=3.2))
print(f"10.5 + 3.2 = {response.result}")
# 减法
response = stub.Subtract(calculator_pb2.SubtractRequest(a=10.5, b=3.2))
print(f"10.5 - 3.2 = {response.result}")
def client_streaming():
"""客户端流式调用"""
channel = grpc.insecure_channel("localhost:50051")
stub = calculator_pb2_grpc.CalculatorStub(channel)
def generate_numbers():
numbers = [1.1, 2.2, 3.3, 4.4, 5.5]
for n in numbers:
yield calculator_pb2.Number(value=n)
response = stub.StreamingAdd(generate_numbers())
print(f"流式求和: 1.1+2.2+3.3+4.4+5.5 = {response.value}")
def server_streaming():
"""服务端流式调用（这里用双向流演示）"""
channel = grpc.insecure_channel("localhost:50051")
stub = calculator_pb2_grpc.CalculatorStub(channel)
def generate_requests():
operations = [
(10, "+", 5),
(10, "-", 5),
(10, "*", 5),
(10, "/", 5),
]
for a, op, b in operations:
yield calculator_pb2.CalcRequest(a=a, op=op, b=b)
responses = stub.StreamingCalc(generate_requests())
for response in responses:
print(f"计算结果: {response.result} (操作: {response.op})")
if __name__ == "__main__":
print("=== 普通RPC调用 ===")
unary_calls()
print("\n=== 客户端流式调用 ===")
client_streaming()
print("\n=== 双向流式调用 ===")
server_streaming()
完整项目实战——构建一个完整的Web应用
7.1 项目结构
http-full-project/
├── app/
│ ├── __init__.py
│ ├── models.py # 数据模型
│ ├── views.py # 路由处理
│ ├── api.py # REST API
│ ├── websocket.py # WebSocket处理
│ └── auth.py # 认证模块
├── templates/
│ ├── index.html
│ ├── login.html
│ └── dashboard.html
├── static/
│ ├── css/
│ │ └── style.css
│ └── js/
│ └── app.js
├── requirements.txt
└── run.py
7.2 应用代码
# app/__init__.py
from flask import Flask
from flask_socketio import SocketIO
from flask_cors import CORS
socketio = SocketIO(cors_allowed_origins="*")
def create_app():
app = Flask(__name__)
app.config["SECRET_KEY"] = "dev-secret-key"
CORS(app)
socketio.init_app(app)
from app.views import main_bp
from app.api import api_bp
from app.websocket import ws_bp
app.register_blueprint(main_bp)
app.register_blueprint(api_bp, url_prefix="/api")
app.register_blueprint(ws_bp, url_prefix="/ws")
return app
# app/models.py
import sqlite3
import datetime
import uuid
from typing import Optional, List, Dict
class Database:
def __init__(self, db_path="app.db"):
self.db_path = db_path
self._init_db()
def _get_conn(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _init_db(self):
conn = self._get_conn()
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
email TEXT,
created_at TEXT NOT NULL,
role TEXT DEFAULT 'user'
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS articles (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL,
author_id TEXT NOT NULL,
category TEXT DEFAULT 'general',
view_count INTEGER DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY (author_id) REFERENCES users(id)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS comments (
id TEXT PRIMARY KEY,
article_id TEXT NOT NULL,
user_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY (article_id) REFERENCES articles(id),
FOREIGN KEY (user_id) REFERENCES users(id)
)
""")
conn.commit()
conn.close()
# ===== 用户操作 =====
def create_user(self, username: str, password_hash: str, email: str = None, role: str = "user") -> str:
user_id = str(uuid.uuid4())
now = datetime.datetime.utcnow().isoformat() + "Z"
conn = self._get_conn()
try:
conn.execute(
"INSERT INTO users (id, username, password_hash, email, created_at, role) VALUES (?,?,?,?,?,?)",
(user_id, username, password_hash, email, now, role)
)
conn.commit()
return user_id
except sqlite3.IntegrityError:
return None
finally:
conn.close()
def find_user_by_username(self, username: str) -> Optional[Dict]:
conn = self._get_conn()
row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
conn.close()
return dict(row) if row else None
# ===== 文章操作 =====
def create_article(self, title: str, content: str, author_id: str, category: str = "general") -> str:
article_id = str(uuid.uuid4())
now = datetime.datetime.utcnow().isoformat() + "Z"
conn = self._get_conn()
conn.execute(
"INSERT INTO articles (id, title, content, author_id, category, created_at, updated_at) VALUES (?,?,?,?,?,?,?)",
(article_id, title, content, author_id, category, now, now)
)
conn.commit()
conn.close()
return article_id
def get_articles(self, page: int = 1, page_size: int = 10, category: str = None) -> Dict:
conn = self._get_conn()
if category:
total = conn.execute(
"SELECT COUNT(*) FROM articles WHERE category = ?", (category,)
).fetchone()[0]
articles = conn.execute(
"SELECT * FROM articles WHERE category = ? ORDER BY created_at DESC LIMIT ? OFFSET ?",
(category, page_size, (page - 1) * page_size)
).fetchall()
else:
total = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
articles = conn.execute(
"SELECT * FROM articles ORDER BY created_at DESC LIMIT ? OFFSET ?",
(page_size, (page - 1) * page_size)
).fetchall()
conn.close()
return {
"items": [dict(a) for a in articles],
"pagination": {
"page": page,
"page_size": page_size,
"total": total,
"total_pages": (total + page_size - 1) // page_size
}
}
def get_article(self, article_id: str) -> Optional[Dict]:
conn = self._get_conn()
row = conn.execute("SELECT * FROM articles WHERE id = ?", (article_id,)).fetchone()
conn.close()
return dict(row) if row else None
def increment_view(self, article_id: str):
conn = self._get_conn()
conn.execute("UPDATE articles SET view_count = view_count + 1 WHERE id = ?", (article_id,))
conn.commit()
conn.close()
# ===== 评论操作 =====
def add_comment(self, article_id: str, user_id: str, content: str) -> str:
comment_id = str(uuid.uuid4())
now = datetime.datetime.utcnow().isoformat() + "Z"
conn = self._get_conn()
conn.execute(
"INSERT INTO comments (id, article_id, user_id, content, created_at) VALUES (?,?,?,?,?)",
(comment_id, article_id, user_id, content, now)
)
conn.commit()
conn.close()
return comment_id
def get_comments(self, article_id: str) -> List[Dict]:
conn = self._get_conn()
rows = conn.execute(
"SELECT * FROM comments WHERE article_id = ? ORDER BY created_at ASC",
(article_id,)
).fetchall()
conn.close()
return [dict(r) for r in rows]
# app/api.py
from flask import Blueprint, jsonify, request, abort
from werkzeug.security import generate_password_hash, check_password_hash
import jwt
import datetime
from app import socketio
from app.models import Database
api_bp = Blueprint("api", __name__)
db = Database()
SECRET_KEY = "your-secret-key"
def auth_required(f):
def wrapper(*args, **kwargs):
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
abort(401)
token = auth_header[7:]
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
request.user = payload
except:
abort(401)
return f(*args, **kwargs)
return wrapper
@api_bp.route("/register", methods=["POST"])
def register():
data = request.get_json()
if not data or not data.get("username") or not data.get("password"):
abort(400)
username = data["username"]
password_hash = generate_password_hash(data["password"])
email = data.get("email")
user_id = db.create_user(username, password_hash, email)
if not user_id:
abort(409, description="用户名已存在")
return jsonify({"code": 201, "message": "注册成功", "data": {"user_id": user_id}}), 201
@api_bp.route("/login", methods=["POST"])
def login():
data = request.get_json()
user = db.find_user_by_username(data.get("username"))
if not user or not check_password_hash(user["password_hash"], data.get("password")):
abort(401, description="用户名或密码错误")
token = jwt.encode({
"user_id": user["id"],
"username": user["username"],
"role": user["role"],
"exp": datetime.datetime.utcnow() + datetime.timedelta(hours=24)
}, SECRET_KEY, algorithm="HS256")
return jsonify({"code": 200, "data": {"token": token, "user": {"id": user["id"], "username": user["username"]}}})
@api_bp.route("/articles", methods=["GET"])
def list_articles():
page = request.args.get("page", 1, type=int)
page_size = request.args.get("page_size", 10, type=int)
category = request.args.get("category")
result = db.get_articles(page, page_size, category)
return jsonify({"code": 200, "data": result})
@api_bp.route("/articles", methods=["POST"])
@auth_required
def create_article():
data = request.get_json()
if not data or not data.get("title") or not data.get("content"):
abort(400)
article_id = db.create_article(
data["title"], data["content"],
request.user["user_id"],
data.get("category", "general")
)
return jsonify({"code": 201, "data": {"id": article_id}}), 201
@api_bp.route("/articles/<article_id>", methods=["GET"])
def get_article(article_id):
article = db.get_article(article_id)
if not article:
abort(404)
db.increment_view(article_id)
return jsonify({"code": 200, "data": article})
@api_bp.route("/articles/<article_id>/comments", methods=["POST"])
@auth_required
def add_comment(article_id):
if not db.get_article(article_id):
abort(404)
data = request.get_json()
if not data or not data.get("content"):
abort(400)
comment_id = db.add_comment(article_id, request.user["user_id"], data["content"])
return jsonify({"code": 201, "data": {"id": comment_id}}), 201
@api_bp.route("/articles/<article_id>/comments", methods=["GET"])
def get_comments(article_id):
comments = db.get_comments(article_id)
return jsonify({"code": 200, "data": comments})
# run.py
from app import create_app, socketio
app = create_app()
if __name__ == "__main__":
print("🚀 服务器启动中...")
print("📍 访问地址: http://localhost:5000")
socketio.run(app, host="0.0.0.0", port=5000, debug=True)
测试你的HTTP服务——Postman和curl实战
8.1 用curl测试API
# 注册
curl -X POST http://localhost:5000/api/register \
-H "Content-Type: application/json" \
-d '{"username": "testuser", "password": "test123", "email": "test@example.com"}'
# 登录获取Token
TOKEN=$(curl -s -X POST http://localhost:5000/api/login \
-H "Content-Type: application/json" \
-d '{"username": "testuser", "password": "test123"}' | \
jq -r '.data.token')
echo "Token: $TOKEN"
# 创建文章
curl -X POST http://localhost:5000/api/articles \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" \
-d '{"title": "我的第一篇文章", "content": "Hello World!", "category": "tech"}'
# 获取文章列表
curl http://localhost:5000/api/articles
# 获取单篇文章
curl http://localhost:5000/api/articles/$(curl http://localhost:5000/api/articles | jq -r '.data.items[0].id')
# 添加评论
curl -X POST http://localhost:5000/api/articles/<article_id>/comments \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" \
-d '{"content": "写得真好！"}'
8.2 用Python自动化测试
import requests
import unittest
class TestAPI(unittest.TestCase):
def setUp(self):
self.base_url = "http://localhost:5000/api"
self.session = requests.Session()
self.token = None
def login(self):
response = self.session.post(f"{self.base_url}/login", json={
"username": "testuser",
"password": "test123"
})
self.token = response.json()["data"]["token"]
self.session.headers["Authorization"] = f"Bearer {self.token}"
def test_register_and_login(self):
# 注册
response = self.session.post(f"{self.base_url}/register", json={
"username": "unittest_user",
"password": "unittest123",
"email": "test@example.com"
})
self.assertEqual(response.status_code, 201)
# 登录
self.login()
self.assertIsNotNone(self.token)
def test_create_and_get_article(self):
self.login()
# 创建文章
response = self.session.post(f"{self.base_url}/articles", json={
"title": "测试文章",
"content": "这是测试内容",
"category": "test"
})
self.assertEqual(response.status_code, 201)
article_id = response.json()["data"]["id"]
# 获取文章
response = self.session.get(f"{self.base_url}/articles/{article_id}")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["data"]["title"], "测试文章")
def test_add_comment(self):
self.login()
# 先创建文章
response = self.session.post(f"{self.base_url}/articles", json={
"title": "评论测试",
"content": "测试内容"
})
article_id = response.json()["data"]["id"]
# 添加评论
response = self.session.post(f"{self.base_url}/articles/{article_id}/comments", json={
"content": "这是条评论"
})
self.assertEqual(response.status_code, 201)
# 获取评论列表
response = self.session.get(f"{self.base_url}/articles/{article_id}/comments")
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.json()["data"]), 1)
def test_unauthorized_access(self):
response = self.session.get(f"{self.base_url}/articles")
self.assertEqual(response.status_code, 401)
if __name__ == "__main__":
unittest.main()
性能优化和最佳实践
9.1 连接池和会话复用
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 创建带连接池的Session
session = requests.Session()
# 配置重试策略
retry_strategy = Retry(
total=3, # 最大重试次数
backoff_factor=1, # 重试间隔: 1s, 2s, 4s
status_forcelist=[429, 500, 502, 503, 504] # 需要重试的状态码
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
# 复用连接，性能提升显著
urls = [
"https://httpbin.org/get",
"https://httpbin.org/headers",
"https://httpbin.org/ip",
]
for url in urls:
response = session.get(url)
print(f"{url}: {response.status_code}")
9.2 异步并发请求
import asyncio
import aiohttp
async def fetch_session(session, url):
"""异步获取单个URL"""
async with session.get(url) as response:
return {
"url": url,
"status": response.status,
"content_length": len(await response.text())
}
async def main():
urls = [
"https://httpbin.org/get",
"https://httpbin.org/headers",
"https://httpbin.org/ip",
"https://httpbin.org/user-agent",
"https://httpbin.org/delay/1",
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_session(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
print(f"错误: {result}")
else:
print(f"✅ {result['url']}: {result['status']} ({result['content_length']} bytes)")
asyncio.run(main())
9.3 请求限流——保护你的爬虫不被封
import asyncio
import aiohttp
import time
class RateLimiter:
"""令牌桶限流器"""
def __init__(self, rate: float, burst: int = 10):
"""
rate: 每秒允许的请求数
burst: 最大突发请求数
"""
self.rate = rate
self.burst = burst
self.tokens = burst
self.last_time = time.monotonic()
async def acquire(self):
"""获取令牌，等待直到有可用令牌"""
while True:
now = time.monotonic()
elapsed = now - self.last_time
self.tokens = min(self.burst, self.tokens + elapsed * self.rate)
self.last_time = now
if self.tokens >= 1:
self.tokens -= 1
return
# 计算等待时间
wait_time = (1 - self.tokens) / self.rate
await asyncio.sleep(wait_time)
async def crawl_with_rate_limit(urls: list, rate: float = 5.0):
"""带限流的并发爬取"""
limiter = RateLimiter(rate=rate)
results = []
async with aiohttp.ClientSession() as session:
async def fetch(url):
await limiter.acquire()
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
if resp.status == 200:
return {"url": url, "status": 200}
else:
return {"url": url, "status": resp.status}
except Exception as e:
return {"url": url, "error": str(e)}
tasks = [fetch(url) for url in urls]
results = await asyncio.gather(*tasks)
return results
# 使用示例
# urls = ["https://httpbin.org/get"] * 20
# results = asyncio.run(crawl_with_rate_limit(urls, rate=5.0))
# print(f"完成 {len(results)} 个请求")
常见问题排查指南
10.1 常见错误及解决方案
import requests
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# ===== 问题1: ConnectionError =====
# 原因: 网络不通、DNS解析失败、服务器拒绝连接
def handle_connection_error():
try:
response = requests.get("https://example.com", timeout=5)
except requests.exceptions.ConnectionError as e:
logger.error(f"连接错误: {e}")
# 解决方案:
# 1. 检查网络连接
# 2. 检查URL是否正确
# 3. 检查是否需要代理
# 4. 检查服务器是否在线
# ===== 问题2: Timeout =====
try:
response = requests.get("https://example.com/slow-endpoint", timeout=2)
except requests.exceptions.Timeout as e:
logger.error(f"超时: {e}")
# 解决方案:
# 1. 增加timeout值
# 2. 检查服务器负载
# 3. 使用异步请求提高效率
# ===== 问题3: HTTP错误 =====
def handle_http_error():
response = requests.get("https://httpbin.org/status/404")
# 方式一: 手动检查
if response.status_code == 404:
logger.warning("页面不存在")
elif response.status_code == 403:
logger.warning("权限不足，可能需要认证")
elif response.status_code == 500:
logger.error("服务器内部错误")
# 方式二: raise_for_status() 自动抛出异常
try:
response.raise_for_status()
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP错误: {e}")
# ===== 问题4: 编码问题 =====
def handle_encoding():
response = requests.get("https://example.com")
# 自动检测编码
response.encoding = response.apparent_encoding
logger.info(f"检测到的编码: {response.encoding}")
# 或者手动设置
response.encoding = "utf-8"
# ===== 问题5: JSON解析失败 =====
def handle_json_error():
response = requests.get("https://httpbin.org/get")
try:
data = response.json()
except requests.exceptions.JSONDecodeError as e:
logger.error(f"JSON解析失败: {e}")
# 检查响应内容类型
content_type = response.headers.get("Content-Type", "")
logger.info(f"Content-Type: {content_type}")
# 如果不是JSON，尝试其他解析方式
if "text/html" in content_type:
logger.warning("返回的是HTML，不是JSON！")
handle_connection_error()
handle_http_error()
handle_encoding()
handle_json_error()
10.2 调试技巧
import requests
from requests.adapters import HTTPAdapter
# 开启详细日志
import logging
http_logger = logging.getLogger("urllib3")
http_logger.setLevel(logging.DEBUG)
http_logger.addHandler(logging.StreamHandler())
# 创建调试用的Session
class DebugSession(requests.Session):
def send(self, request, **kwargs):
print(f"\n📤 请求: {request.method} {request.url}")
print(f"📋 Headers: {dict(request.headers)}")
if request.body:
print(f"📦 Body: {request.body[:200]}")
response = super().send(request, **kwargs)
print(f"\n📥 响应: {response.status_code}")
print(f"📋 Response Headers: {dict(response.headers)}")
print(f"📦 Response Body: {response.text[:200]}...")
return response
# 使用调试Session
debug_session = DebugSession()
response = debug_session.get("https://httpbin.org/get", params={"test": "value"})
好了，这篇东西挺长的，从最基础的requests用法，到爬虫实战，到服务器搭建，再到WebSocket和gRPC，基本上把HTTP编程的常见场景都覆盖了。我写这篇的时候一直在想，要是当初我学这些东西的时候有人能这么系统地给我讲一遍，该多好。
记住几个核心要点：HTTP就是请求和响应，状态码是沟通的语言，安全要放在第一位。代码多敲几遍，遇到问题别慌，先看状态码，再查日志，大部分问题都能解决。有什么具体问题随时问我，咱们一起把HTTP这个工具用得越来越顺手。