路由与视图¶
📖 章节简介¶
本章将深入讲解Flask的路由系统和视图函数,这是Flask应用的核心部分。你将学习如何定义路由、处理HTTP请求、以及构建响应。
🛣️ 路由基础¶
1. 路由定义¶
基本路由¶
Python
from flask import Flask, render_template
app = Flask(__name__)
# 基本路由
@app.route('/')
def index():
return '首页'
@app.route('/hello')
def hello():
return '你好,世界!'
# 带参数的路由
@app.route('/user/<username>')
def show_user_profile(username):
return f'用户: {username}'
# 带类型限制的路由
@app.route('/post/<int:post_id>')
def show_post(post_id):
return f'文章ID: {post_id}'
@app.route('/path/<path:subpath>')
def show_subpath(subpath):
return f'子路径: {subpath}'
# 带浮点数的路由
@app.route('/price/<float:price>')
def show_price(price):
return f'价格: ¥{price:.2f}'
路由转换器¶
Python
# Flask内置的转换器
converters = {
'string': '默认类型,接受任何不包含斜杠的文本',
'int': '接受正整数',
'float': '接受正浮点数',
'path': '类似string,但接受斜杠',
'uuid': '接受UUID字符串'
}
# 自定义转换器
from werkzeug.routing import BaseConverter
class ListConverter(BaseConverter):
"""自定义列表转换器"""
def to_python(self, value):
return value.split(',')
def to_url(self, values):
return ','.join(super().to_url(value) for value in values) # super()调用父类方法
# 注册自定义转换器
app.url_map.converters['list'] = ListConverter
# 使用自定义转换器
@app.route('/tags/<list:tags>')
def show_tags(tags):
return f'标签: {", ".join(tags)}'
2. HTTP方法¶
支持的HTTP方法¶
Python
from flask import request
# 指定允许的HTTP方法
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
return '处理登录请求'
return '显示登录表单'
# RESTful风格的路由
@app.route('/api/users', methods=['GET', 'POST'])
def users():
if request.method == 'GET':
return '获取用户列表'
elif request.method == 'POST':
return '创建新用户'
@app.route('/api/users/<int:user_id>', methods=['GET', 'PUT', 'DELETE'])
def user_detail(user_id):
if request.method == 'GET':
return f'获取用户 {user_id}'
elif request.method == 'PUT':
return f'更新用户 {user_id}'
elif request.method == 'DELETE':
return f'删除用户 {user_id}'
HTTP方法示例¶
Python
from flask import jsonify
# GET - 获取资源
@app.route('/api/products', methods=['GET'])
def get_products():
products = [
{'id': 1, 'name': '产品A', 'price': 100},
{'id': 2, 'name': '产品B', 'price': 200}
]
return jsonify(products)
# POST - 创建资源
@app.route('/api/products', methods=['POST'])
def create_product():
data = request.get_json()
# 创建产品的逻辑
return jsonify({'message': '产品创建成功', 'product': data}), 201
# PUT - 更新资源
@app.route('/api/products/<int:product_id>', methods=['PUT'])
def update_product(product_id):
data = request.get_json()
# 更新产品的逻辑
return jsonify({'message': f'产品 {product_id} 更新成功'})
# DELETE - 删除资源
@app.route('/api/products/<int:product_id>', methods=['DELETE'])
def delete_product(product_id):
# 删除产品的逻辑
return jsonify({'message': f'产品 {product_id} 删除成功'})
📋 请求处理¶
1. 请求对象¶
获取请求数据¶
Python
from flask import request
@app.route('/request-info')
def request_info():
# 获取请求方法
method = request.method
# 获取URL参数
name = request.args.get('name', '默认值')
# 获取表单数据
username = request.form.get('username')
password = request.form.get('password')
# 获取JSON数据
data = request.get_json()
# 获取请求头
user_agent = request.headers.get('User-Agent')
# 获取客户端IP
ip = request.remote_addr
# 获取上传的文件
file = request.files.get('file')
return {
'method': method,
'name': name,
'username': username,
'data': data,
'user_agent': user_agent,
'ip': ip
}
文件上传¶
Python
import os
from werkzeug.utils import secure_filename
# 配置上传目录
UPLOAD_FOLDER = 'uploads'
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
def allowed_file(filename):
"""检查文件扩展名是否允许"""
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
if request.method == 'POST':
# 检查是否有文件
if 'file' not in request.files:
return '没有文件', 400
file = request.files['file']
# 检查文件名是否为空
if file.filename == '':
return '没有选择文件', 400
# 检查文件类型
if file and allowed_file(file.filename):
# 安全的文件名
filename = secure_filename(file.filename)
# 保存文件
file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
return f'文件 {filename} 上传成功'
return '''
<!doctype html>
<html>
<head><title>上传文件</title></head>
<body>
<h1>上传文件</h1>
<form method=post enctype=multipart/form-data>
<input type=file name=file>
<input type=submit value=上传>
</form>
</body>
</html>
'''
2. 请求验证¶
数据验证¶
Python
from flask import abort
@app.route('/user/<int:user_id>')
def get_user(user_id):
# 验证用户ID
if user_id <= 0:
abort(400, description='无效的用户ID')
# 查询用户
user = db.session.get(User, user_id)
if not user:
abort(404, description='用户不存在')
return jsonify(user.to_dict())
# 错误处理
@app.errorhandler(400)
def bad_request(error):
return jsonify({'error': error.description}), 400
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': error.description}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({'error': '服务器内部错误'}), 500
📤 响应处理¶
1. 响应类型¶
基本响应¶
Python
from flask import Response, make_response, redirect, url_for
# 返回字符串
@app.route('/text')
def text_response():
return '这是一个文本响应'
# 返回HTML
@app.route('/html')
def html_response():
return '<h1>这是一个HTML响应</h1>'
# 返回JSON
@app.route('/json')
def json_response():
data = {'message': 'Hello', 'status': 'success'}
return jsonify(data)
# 返回元组(响应体,状态码,响应头)
@app.route('/tuple')
def tuple_response():
return '响应内容', 201, {'X-Custom-Header': 'Custom Value'}
# 使用make_response
@app.route('/make-response')
def make_response_example():
resp = make_response('自定义响应')
resp.status_code = 200
resp.headers['X-Custom'] = 'Value'
return resp
# 重定向
@app.route('/redirect')
def redirect_example():
return redirect(url_for('index'))
# 返回Response对象
@app.route('/response')
def response_example():
resp = Response('自定义响应内容', mimetype='text/plain')
return resp
模板渲染¶
Python
from flask import render_template, render_template_string
# 渲染模板
@app.route('/')
def index():
return render_template('index.html', title='首页')
# 传递变量到模板
@app.route('/user/<username>')
def user_profile(username):
user = {
'name': username,
'email': f'{username}@example.com',
'age': 25
}
return render_template('user.html', user=user)
# 渲染模板字符串
@app.route('/inline')
def inline_template():
template = '''
<!DOCTYPE html>
<html>
<head><title>{{ title }}</title></head>
<body>
<h1>{{ title }}</h1>
<p>{{ content }}</p>
</body>
</html>
'''
return render_template_string(template, title='内联模板', content='这是内联渲染的内容')
2. Cookie处理¶
设置和读取Cookie¶
Python
from flask import make_response
# 设置Cookie
@app.route('/set-cookie')
def set_cookie():
resp = make_response('Cookie已设置')
resp.set_cookie('username', '张三', max_age=3600)
resp.set_cookie('user_id', '12345', max_age=3600)
return resp
# 读取Cookie
@app.route('/get-cookie')
def get_cookie():
username = request.cookies.get('username')
user_id = request.cookies.get('user_id')
return f'用户名: {username}, 用户ID: {user_id}'
# 删除Cookie
@app.route('/delete-cookie')
def delete_cookie():
resp = make_response('Cookie已删除')
resp.delete_cookie('username')
resp.delete_cookie('user_id')
return resp
# 安全的Cookie(需要设置SECRET_KEY)
@app.route('/set-secure-cookie')
def set_secure_cookie():
resp = make_response('安全Cookie已设置')
resp.set_cookie('session_id', 'abc123',
secure=True, # 仅HTTPS
httponly=True, # 禁止JavaScript访问
samesite='Lax') # 防止CSRF
return resp
3. Session处理¶
Session基本使用¶
Python
from flask import session
from functools import wraps
# 设置SECRET_KEY
app.secret_key = 'your-secret-key-here'
# 设置Session
@app.route('/login', methods=['POST'])
def login():
username = request.form.get('username')
password = request.form.get('password')
# 验证用户
if authenticate(username, password):
session['user_id'] = user.id
session['username'] = username
return '登录成功'
return '登录失败'
# 读取Session
@app.route('/profile')
def profile():
if 'user_id' in session:
user_id = session['user_id']
username = session.get('username')
return f'用户ID: {user_id}, 用户名: {username}'
return '请先登录'
# 删除Session
@app.route('/logout')
def logout():
session.clear()
return '已登出'
# 检查用户是否登录
def login_required(f):
"""登录验证装饰器"""
@wraps(f)
def decorated_function(*args, **kwargs): # *args接收任意位置参数;**kwargs接收任意关键字参数
if 'user_id' not in session:
return redirect(url_for('login'))
return f(*args, **kwargs)
return decorated_function
@app.route('/protected')
@login_required
def protected():
return '这是一个需要登录的页面'
🎯 蓝图(Blueprint)¶
1. 蓝图基础¶
创建蓝图¶
Python
# app/main/__init__.py
from flask import Blueprint
main_bp = Blueprint('main', __name__)
from app.main import routes
Python
# app/main/routes.py
from flask import render_template, Blueprint
from app.main import main_bp
@main_bp.route('/')
def index():
return render_template('index.html')
@main_bp.route('/about')
def about():
return render_template('about.html')
Python
# app/auth/__init__.py
from flask import Blueprint
auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
from app.auth import routes
Python
# app/auth/routes.py
from flask import render_template, Blueprint
from app.auth import auth_bp
@auth_bp.route('/login')
def login():
return render_template('auth/login.html')
@auth_bp.route('/register')
def register():
return render_template('auth/register.html')
2. 注册蓝图¶
Python
# app/__init__.py
from flask import Flask
def create_app():
app = Flask(__name__)
# 注册蓝图
from app.main import bp as main_bp
app.register_blueprint(main_bp)
from app.auth import bp as auth_bp
app.register_blueprint(auth_bp, url_prefix='/auth')
return app
💡 最佳实践¶
1. 路由组织¶
Python
# 推荐的路由组织方式
route_organization = {
'使用蓝图': '将相关路由组织到蓝图中',
'RESTful设计': '遵循RESTful API设计原则',
'命名规范': '使用有意义的路由名称',
'版本控制': 'API使用版本号',
'文档化': '为API编写文档'
}
2. 错误处理¶
Python
# 统一错误处理
@app.errorhandler(Exception)
def handle_exception(e):
"""处理所有未捕获的异常"""
if isinstance(e, HTTPException): # isinstance检查对象类型
return jsonify({'error': e.description}), e.code
# 记录错误日志
app.logger.error(f'未处理的异常: {str(e)}')
return jsonify({'error': '服务器内部错误'}), 500
# 自定义异常
class ValidationError(Exception):
"""验证错误"""
pass
@app.errorhandler(ValidationError)
def handle_validation_error(e):
return jsonify({'error': str(e)}), 400
📝 练习题¶
基础题¶
- 什么是路由?如何定义Flask路由?
- Flask支持哪些HTTP方法?
- 如何获取请求参数?
进阶题¶
- 实现一个RESTful API。
- 如何处理文件上传?
- 使用蓝图组织你的应用。
实践题¶
- 创建一个用户管理API。
- 实现文件上传功能。
- 构建一个完整的CRUD应用。
📚 推荐阅读¶
🔗 下一章¶
模板引擎 - 学习Jinja2模板引擎的使用。