Python 爬虫技术 第06节 HTTP协议与Web基础知识
HTTP（Hypertext Transfer Protocol）是用于从Web服务器传输超文本到本地浏览器的传输协议。它是互联网上应用最为广泛的一种网络协议，几乎所有的网页数据都是通过HTTP协议进行传输的。下面，我将结合一个简单的Python案例来详细讲解HTTP协议与Web基础知识。
HTTP协议基础
请求与响应模型
HTTP遵循请求/响应模型。客户端（通常是浏览器）发送一个HTTP请求到服务器，服务器收到请求后，会返回一个HTTP响应。
请求结构
一个典型的HTTP请求由三部分组成：
- 请求行 - 包含方法（GET, POST, PUT, DELETE等）、资源的URL和使用的HTTP版本。
- 请求头 - 包含客户端信息、认证信息、编码偏好、缓存控制等。
- 请求体 - 在POST、PUT等请求中，包含要发送的数据。
响应结构
HTTP响应也由三部分组成：
- 状态行 - 包括HTTP版本、状态码和描述状态的短语。
- 响应头 - 包含服务器信息、缓存控制、日期时间、内容类型等。
- 响应体 - 实际的响应数据，例如HTML文档、图片、视频等。
Python案例：使用http.client模块发送HTTP请求
http.client是Python标准库的一部分，提供了发送HTTP请求的功能。下面是一个使用http.client发送GET请求的例子：
import http.client
import json
# 定义主机和端口
host = "www.example.com"
port = 80
# 创建连接
conn = http.client.HTTPConnection(host, port)
# 发送GET请求
conn.request("GET", "/")
# 获取响应
response = conn.getresponse()
# 打印状态码
print("Status:", response.status, response.reason)
# 读取响应数据
data = response.read().decode()
# 打印响应数据
print("Data:\n", data)
# 关闭连接
conn.close()
Python案例：使用requests库发送HTTP请求
requests库是一个非常流行的第三方库，用于发送HTTP请求。它比http.client更高级，更易用，支持更多功能。
import requests
# 发送GET请求
response = requests.get("http://www.example.com")
# 检查响应状态码
if response.status_code == 200:
# 打印响应内容
print(response.text)
else:
print("Request failed with status code:", response.status_code)
Python案例：使用Flask框架创建Web服务器
Flask是一个轻量级的Web框架，非常适合用于构建小型Web应用程序或API服务。下面是一个简单的Flask应用示例：
from flask import Flask
app = Flask(__name__)
@app.route('/')
def home():
return "Hello, World!"
if __name__ == '__main__':
app.run(debug=True)
在这个例子中，我们创建了一个简单的Web服务器，它监听所有外部接口上的默认端口（5000），并在接收到请求时返回"Hello, World!"的消息。
总结
HTTP协议是Web通信的基础，理解其工作原理对于开发Web应用程序至关重要。Python提供了多种工具和库来处理HTTP请求和响应，无论是客户端还是服务器端编程，都可以轻松实现。通过上述案例，你应该能够开始编写自己的HTTP客户端和服务器代码了。
当我们谈论扩展基于HTTP的Python代码时，可以考虑增加的功能有很多。这里，我将展示如何使用requests库来发送更复杂的HTTP请求（如POST请求），以及如何使用Flask框架来处理这些请求。
使用requests发送POST请求
假设我们有一个Web服务，它期望接收JSON格式的数据并通过POST请求发送。下面是如何使用requests库发送这样一个请求：
import requests
import json
url = "http://localhost:5000/data" # 假设这是你的Flask应用的URL
data = {
"name": "John Doe",
"age": 30,
"city": "New York"
}
headers = {'Content-Type': 'application/json'}
response = requests.post(url, data=json.dumps(data), headers=headers)
if response.status_code == 200:
print("Request successful!")
print("Response:", response.json())
else:
print("Request failed with status code:", response.status_code)
使用Flask接收POST请求
接下来，我们将在Flask应用中添加一个端点来接收上述POST请求：
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/data', methods=['POST'])
def receive_data():
if request.headers['Content-Type'] == 'application/json':
data = request.json
print("Received data:", data)
# 这里可以处理数据，例如保存到数据库
return jsonify({"message": "Data received successfully"}), 200
else:
return jsonify({"error": "Invalid content type"}), 400
if __name__ == '__main__':
app.run(debug=True)
扩展Flask应用
我们还可以为Flask应用添加更多的路由和功能。例如，添加一个GET请求来获取数据：
@app.route('/data', methods=['GET'])
def get_data():
# 假设这里有从数据库获取数据的逻辑
data = {"name": "John Doe", "age": 30, "city": "New York"}
return jsonify(data), 200
错误处理
在Flask中，可以使用错误处理器来处理各种HTTP错误：
@app.errorhandler(404)
def page_not_found(e):
return jsonify({"error": "Resource not found"}), 404
@app.errorhandler(500)
def internal_server_error(e):
return jsonify({"error": "Internal server error"}), 500
使用环境变量和配置
为了提高安全性，可以使用环境变量来管理敏感信息，如数据库URL、API密钥等：
import os
app.config['DATABASE_URL'] = os.getenv('DATABASE_URL')
然后，在部署环境中设置这些环境变量：
export DATABASE_URL=postgres://user:pass@localhost/dbname
使用Flask扩展
Flask有许多扩展，可以简化常见的任务，如数据库集成、身份验证、表单处理等。例如，使用Flask-SQLAlchemy来集成SQLAlchemy ORM：
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
def __repr__(self):
return '<User %r>' % self.username
通过这些扩展和增强功能，你可以构建功能丰富且安全的Web应用程序。这只是一个开始，实际的应用可能涉及更复杂的业务逻辑、安全措施、性能优化和部署策略。
既然我们已经讨论了如何使用requests库来发送POST请求和使用Flask框架来接收和处理这些请求，我们可以进一步完善代码，使其更具实用性。以下是几个方面的扩展：
- 数据库集成：使用
Flask-SQLAlchemy来持久化数据。 - 身份验证：使用
Flask-Login或Flask-JWT来管理用户身份验证。 - 表单验证：使用
WTForms来处理和验证表单数据。 - 日志记录：使用
logging模块来记录重要事件和错误。 - 单元测试：使用
pytest来编写和运行单元测试。
首先，让我们继续使用Flask-SQLAlchemy来管理用户数据：
安装Flask-SQLAlchemy
在终端中安装Flask-SQLAlchemy：
pip install Flask-SQLAlchemy
配置Flask-SQLAlchemy
在Flask应用中添加数据库配置：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db' # 或者使用其他数据库
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
创建数据库模型
定义一个用户模型：
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), nullable=False)
age = db.Column(db.Integer)
city = db.Column(db.String(120))
def __repr__(self):
return f'<User {self.name}>'
创建数据库表
在应用运行前创建数据库表：
if __name__ == '__main__':
db.create_all()
app.run(debug=True)
修改POST路由以保存数据
更新receive_data函数以将数据保存到数据库：
@app.route('/data', methods=['POST'])
def receive_data():
if request.headers['Content-Type'] == 'application/json':
data = request.json
new_user = User(name=data['name'], age=data['age'], city=data['city'])
db.session.add(new_user)
db.session.commit()
return jsonify({"message": "Data received and saved successfully"}), 201
else:
return jsonify({"error": "Invalid content type"}), 400
添加GET路由以查询数据
添加一个路由来查询所有用户数据：
@app.route('/data', methods=['GET'])
def get_data():
users = User.query.all()
return jsonify([user.to_dict() for user in users]), 200
定义to_dict方法
在User类中添加一个方法来方便地将对象转换为字典：
class User(db.Model):
# ...
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'age': self.age,
'city': self.city
}
日志记录
使用Python的logging模块记录关键操作：
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@app.route('/data', methods=['POST'])
def receive_data():
# ...
logger.info(f"New user data received: {data}")
# ...
单元测试
使用pytest编写单元测试：
# tests/test_app.py
import pytest
from app import app, db
from app.models import User
@pytest.fixture
def client():
app.config['TESTING'] = True
client = app.test_client()
with app.app_context():
db.create_all()
yield client
with app.app_context():
db.drop_all()
def test_receive_data(client):
response = client.post('/data', json={
"name": "Test User",
"age": 25,
"city": "Test City"
}, content_type='application/json')
assert response.status_code == 201
assert User.query.count() == 1
以上步骤将使你的应用更加健壮和实用，但请记住，实际部署时可能需要考虑更多的细节，例如生产环境下的数据库配置、安全性、性能优化等。
原文地址：https://blog.csdn.net/hummhumm/article/details/140587804
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！