Python Flask 框架开发指南

前言

Flask 是一个轻量级的 Python Web 应用框架，被设计为易于使用且易于扩展。它以其简洁、灵活的特性而闻名，非常适合从小型应用到复杂的企业级 Web 服务。本文将全面介绍 Flask 框架的核心概念和实践技巧，帮助你快速掌握 Python Web 开发。

基础入门

环境准备

首先，我们需要安装 Flask 及其依赖：

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows
venv\Scripts\activate
# macOS/Linux
source venv/bin/activate

# 安装 Flask
pip install flask

# 将依赖保存到 requirements.txt
pip freeze > requirements.txt

创建第一个应用

创建一个名为 app.py 的文件，包含基本的 Flask 应用：

from flask import Flask

# 创建 Flask 应用实例
app = Flask(__name__)

# 定义路由
@app.route('/')
def hello_world():
    return 'Hello, World!'

# 运行应用
if __name__ == '__main__':
    app.run(debug=True)

运行应用并在浏览器中访问 http://127.0.0.1:5000/：

python app.py

项目结构

随着项目的增长，良好的项目结构变得至关重要：

my_flask_app/
├── app/
│   ├── __init__.py       # 应用工厂
│   ├── routes.py         # 路由定义
│   ├── models.py         # 数据模型
│   ├── forms.py          # 表单类
│   ├── static/           # 静态文件(CSS, JS, 图片)
│   └── templates/        # HTML 模板
├── config.py             # 配置文件
├── requirements.txt      # 依赖清单
└── run.py                # 启动脚本

路由系统

基本路由

Flask 使用装饰器定义路由：

@app.route('/')
def index():
    return 'Index Page'

@app.route('/hello')
def hello():
    return 'Hello, World!'

动态路由

可以在路由中包含变量部分：

@app.route('/user/<username>')
def show_user_profile(username):
    return f'User {username}'

@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'Post {post_id}'

HTTP 方法

默认情况下，路由只响应 GET 请求。可以通过 methods 参数指定支持的 HTTP 方法：

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 处理表单提交
        return do_the_login()
    else:
        # 显示登录表单
        return show_login_form()

URL 构建

使用 url_for() 函数构建 URL：

from flask import url_for

@app.route('/')
def index():
    # 生成 'login' 视图的 URL
    login_url = url_for('login')
    # 生成 'profile' 视图的 URL，带有参数
    user_url = url_for('show_user_profile', username='john')
    return f'Login URL: {login_url}, User Profile URL: {user_url}'

请求和响应

请求对象

Flask 将请求信息封装到 request 对象中：

from flask import request

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')
    # 处理登录逻辑
    return f'Logged in as {username}'

@app.route('/upload', methods=['POST'])
def upload_file():
    uploaded_file = request.files['file']
    if uploaded_file.filename != '':
        # 保存文件
        uploaded_file.save('/path/to/uploads/' + uploaded_file.filename)
    return 'File uploaded successfully'

@app.route('/search')
def search():
    query = request.args.get('q', '')
    return f'Search results for: {query}'

响应对象

Flask 视图函数可以返回多种类型的响应：

from flask import jsonify, make_response, render_template, redirect, url_for

# 返回 JSON
@app.route('/api/data')
def get_data():
    data = {'name': 'John', 'email': 'john@example.com'}
    return jsonify(data)

# 自定义响应
@app.route('/custom')
def custom_response():
    resp = make_response('Custom response', 201)
    resp.headers['X-Something'] = 'A value'
    return resp

# 重定向
@app.route('/redirect')
def redirect_example():
    return redirect(url_for('index'))

模板引擎

Flask 使用 Jinja2 作为模板引擎，可以将 Python 变量传递到 HTML 模板中。

基本用法

from flask import render_template

@app.route('/hello/<name>')
def hello(name=None):
    return render_template('hello.html', name=name)

模板文件 templates/hello.html：

<!doctype html>
<html>
  <head>
    <title>Hello from Flask</title>
  </head>
  <body>
    {% if name %}
      <h1>Hello {{ name }}!</h1>
    {% else %}
      <h1>Hello, World!</h1>
    {% endif %}
  </body>
</html>

模板继承

通过模板继承可以复用页面结构：

基础模板 templates/base.html：

<!doctype html>
<html>
  <head>
    <title>{% block title %}{% endblock %} - My Website</title>
    <link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
  </head>
  <body>
    <nav>
      <ul>
        <li><a href="{{ url_for('index') }}">Home</a></li>
        <li><a href="{{ url_for('about') }}">About</a></li>
      </ul>
    </nav>
    <div class="content">
      {% block content %}{% endblock %}
    </div>
    <footer>
      &copy; 2023 My Website
    </footer>
  </body>
</html>

子模板 templates/index.html：

{% extends 'base.html' %}

{% block title %}Home{% endblock %}

{% block content %}
  <h1>Welcome to my website</h1>
  <p>This is the home page.</p>
{% endblock %}

模板过滤器

Jinja2 提供了多种过滤器用于转换变量：

{{ name|capitalize }}
{{ text|truncate(100) }}
{{ list|join(', ') }}

你也可以定义自定义过滤器：

@app.template_filter('reverse')
def reverse_filter(s):
    return s[::-1]

然后在模板中使用：

{{ "Hello"|reverse }}  <!-- 输出: "olleH" -->

静态文件

Flask 将静态文件存放在 static 文件夹中：

app/
└── static/
    ├── css/
    │   └── style.css
    ├── js/
    │   └── script.js
    └── images/
        └── logo.png

在模板中引用静态文件：

<link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
<script src="{{ url_for('static', filename='js/script.js') }}"></script>
<img src="{{ url_for('static', filename='images/logo.png') }}" alt="Logo">

表单处理

基本表单处理

处理 HTML 表单提交：

from flask import request, render_template, redirect, url_for

@app.route('/login', methods=['GET', 'POST'])
def login():
    error = None
    if request.method == 'POST':
        if valid_login(request.form['username'], request.form['password']):
            return redirect(url_for('success'))
        else:
            error = 'Invalid username/password'
    # 当请求方法为 GET 或表单验证失败时，显示表单
    return render_template('login.html', error=error)

使用 Flask-WTF

Flask-WTF 是一个 Flask 扩展，提供了表单验证和 CSRF 保护：

pip install flask-wtf

定义表单类：

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length

class LoginForm(FlaskForm):
    email = StringField('Email', validators=[DataRequired(), Email()])
    password = PasswordField('Password', validators=[DataRequired(), Length(min=6)])
    submit = SubmitField('Log In')

在视图中使用表单：

@app.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        # 表单验证通过
        email = form.email.data
        password = form.password.data
        # 处理登录逻辑
        return redirect(url_for('index'))
    return render_template('login.html', form=form)

在模板中渲染表单：

<form method="post">
  {{ form.hidden_tag() }}
  <div>
    {{ form.email.label }}
    {{ form.email }}
    {% if form.email.errors %}
      <ul class="errors">
        {% for error in form.email.errors %}
          <li>{{ error }}</li>
        {% endfor %}
      </ul>
    {% endif %}
  </div>
  <div>
    {{ form.password.label }}
    {{ form.password }}
    {% if form.password.errors %}
      <ul class="errors">
        {% for error in form.password.errors %}
          <li>{{ error }}</li>
        {% endfor %}
      </ul>
    {% endif %}
  </div>
  {{ form.submit }}
</form>

数据库集成

Flask 本身不提供数据库功能，但可以与多种数据库交互。

使用 SQLAlchemy (ORM)

Flask-SQLAlchemy 是 Flask 的 SQLAlchemy 扩展：

pip install flask-sqlalchemy

基本配置：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
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)
    posts = db.relationship('Post', backref='author', lazy=True)

    def __repr__(self):
        return f'<User {self.username}>'

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

    def __repr__(self):
        return f'<Post {self.title}>'

# 创建数据库表
with app.app_context():
    db.create_all()

使用模型：

@app.route('/user/<username>')
def user_profile(username):
    user = User.query.filter_by(username=username).first_or_404()
    return render_template('profile.html', user=user)

@app.route('/create_user', methods=['POST'])
def create_user():
    username = request.form.get('username')
    email = request.form.get('email')
    
    user = User(username=username, email=email)
    db.session.add(user)
    db.session.commit()
    
    return redirect(url_for('user_profile', username=username))

使用 MongoDB

使用 PyMongo 扩展连接 MongoDB：

pip install flask-pymongo

基本配置：

from flask import Flask
from flask_pymongo import PyMongo

app = Flask(__name__)
app.config["MONGO_URI"] = "mongodb://localhost:27017/myDatabase"
mongo = PyMongo(app)

@app.route('/add_user', methods=['POST'])
def add_user():
    users = mongo.db.users
    users.insert_one({
        'username': request.form.get('username'),
        'email': request.form.get('email')
    })
    return redirect(url_for('index'))

@app.route('/users')
def get_users():
    users = mongo.db.users.find()
    return render_template('users.html', users=users)

身份验证

Flask-Login 是一个用于处理用户会话管理的扩展：

pip install flask-login

基本设置：

from flask import Flask, render_template, redirect, url_for, request, flash
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from werkzeug.security import generate_password_hash, check_password_hash

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login'

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(100), unique=True, nullable=False)
    email = db.Column(db.String(100), unique=True, nullable=False)
    password_hash = db.Column(db.String(200), nullable=False)

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

@app.route('/login', methods=['GET', 'POST'])
def login():
    if current_user.is_authenticated:
        return redirect(url_for('index'))
    
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        
        user = User.query.filter_by(username=username).first()
        if user and user.check_password(password):
            login_user(user)
            next_page = request.args.get('next')
            return redirect(next_page or url_for('index'))
        flash('Invalid username or password')
    
    return render_template('login.html')

@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('index'))

@app.route('/profile')
@login_required
def profile():
    return render_template('profile.html', user=current_user)

蓝图和模块化

随着应用规模增长，可以使用蓝图将应用拆分为可重用的组件：

# app/auth/routes.py
from flask import Blueprint, render_template, redirect, url_for

auth_bp = Blueprint('auth', __name__)

@auth_bp.route('/login')
def login():
    return render_template('auth/login.html')

@auth_bp.route('/register')
def register():
    return render_template('auth/register.html')

# app/main/routes.py
from flask import Blueprint, render_template

main_bp = Blueprint('main', __name__)

@main_bp.route('/')
def index():
    return render_template('main/index.html')

@main_bp.route('/about')
def about():
    return render_template('main/about.html')

# app/__init__.py
from flask import Flask

def create_app():
    app = Flask(__name__)
    
    from app.auth.routes import auth_bp
    from app.main.routes import main_bp
    
    app.register_blueprint(auth_bp, url_prefix='/auth')
    app.register_blueprint(main_bp)
    
    return app

RESTful API 开发

Flask 可以很容易地用于构建 RESTful API：

from flask import Flask, jsonify, request, abort

app = Flask(__name__)

# 示例数据
ITEMS = [
    {'id': 1, 'name': 'Item 1'},
    {'id': 2, 'name': 'Item 2'}
]

# 获取所有项目
@app.route('/api/items', methods=['GET'])
def get_items():
    return jsonify({'items': ITEMS})

# 获取单个项目
@app.route('/api/items/<int:item_id>', methods=['GET'])
def get_item(item_id):
    item = next((item for item in ITEMS if item['id'] == item_id), None)
    if item is None:
        abort(404)
    return jsonify({'item': item})

# 创建新项目
@app.route('/api/items', methods=['POST'])
def create_item():
    if not request.json or 'name' not in request.json:
        abort(400)
    
    item = {
        'id': ITEMS[-1]['id'] + 1 if ITEMS else 1,
        'name': request.json['name']
    }
    ITEMS.append(item)
    return jsonify({'item': item}), 201

# 更新项目
@app.route('/api/items/<int:item_id>', methods=['PUT'])
def update_item(item_id):
    item = next((item for item in ITEMS if item['id'] == item_id), None)
    if item is None:
        abort(404)
    if not request.json:
        abort(400)
    
    item['name'] = request.json.get('name', item['name'])
    return jsonify({'item': item})

# 删除项目
@app.route('/api/items/<int:item_id>', methods=['DELETE'])
def delete_item(item_id):
    item = next((item for item in ITEMS if item['id'] == item_id), None)
    if item is None:
        abort(404)
    ITEMS.remove(item)
    return jsonify({'result': True})

使用 Flask-RESTful 扩展可以更方便地构建 RESTful API：

pip install flask-restful

from flask import Flask
from flask_restful import Api, Resource, reqparse, abort

app = Flask(__name__)
api = Api(app)

# 示例数据
ITEMS = {
    1: {'name': 'Item 1'},
    2: {'name': 'Item 2'}
}

# 请求解析器
parser = reqparse.RequestParser()
parser.add_argument('name', type=str, required=True, help='Name is required')

class ItemResource(Resource):
    def get(self, item_id):
        if item_id not in ITEMS:
            abort(404, message=f"Item {item_id} not found")
        return {'id': item_id, 'item': ITEMS[item_id]}
    
    def put(self, item_id):
        if item_id not in ITEMS:
            abort(404, message=f"Item {item_id} not found")
        args = parser.parse_args()
        ITEMS[item_id] = {'name': args['name']}
        return {'id': item_id, 'item': ITEMS[item_id]}
    
    def delete(self, item_id):
        if item_id not in ITEMS:
            abort(404, message=f"Item {item_id} not found")
        del ITEMS[item_id]
        return '', 204

class ItemListResource(Resource):
    def get(self):
        return {'items': {id: item for id, item in ITEMS.items()}}
    
    def post(self):
        args = parser.parse_args()
        item_id = max(ITEMS.keys()) + 1 if ITEMS else 1
        ITEMS[item_id] = {'name': args['name']}
        return {'id': item_id, 'item': ITEMS[item_id]}, 201

# 添加资源到 API
api.add_resource(ItemResource, '/api/items/<int:item_id>')
api.add_resource(ItemListResource, '/api/items')

部署

使用 Gunicorn

在生产环境中，应使用生产级 WSGI 服务器如 Gunicorn：

pip install gunicorn

启动应用：

gunicorn -w 4 -b 0.0.0.0:8000 'app:create_app()'

使用 Docker

Dockerfile 示例：

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

ENV FLASK_APP=app.py
ENV FLASK_ENV=production

EXPOSE 5000

CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]

构建和运行 Docker 容器：

docker build -t my-flask-app .
docker run -p 5000:5000 my-flask-app

测试

Flask 提供了测试客户端，方便进行单元测试：

import unittest
from app import create_app, db

class FlaskTest(unittest.TestCase):
    def setUp(self):
        self.app = create_app('testing')
        self.client = self.app.test_client()
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
    
    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()
    
    def test_home_page(self):
        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'Welcome', response.data)
    
    def test_create_user(self):
        response = self.client.post('/users/create', data={
            'username': 'testuser',
            'email': 'test@example.com',
            'password': 'password123'
        })
        self.assertEqual(response.status_code, 302)  # 重定向状态码
        
        user = User.query.filter_by(username='testuser').first()
        self.assertIsNotNone(user)
        self.assertEqual(user.email, 'test@example.com')

if __name__ == '__main__':
    unittest.main()

常见扩展

Flask 有丰富的扩展生态系统：

Flask-Migrate: 数据库迁移
Flask-Mail: 发送电子邮件
Flask-Caching: 缓存支持
Flask-Cors: 处理跨域资源共享
Flask-Admin: 管理界面
Flask-Security: 提供常见安全机制
Flask-SocketIO: WebSocket 支持

示例：使用 Flask-Migrate 进行数据库迁移

pip install flask-migrate

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)
migrate = Migrate(app, db)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))
    # 假设我们想添加一个新列
    email = db.Column(db.String(128))

初始化和管理迁移：

flask db init      # 初始化迁移存储库
flask db migrate   # 生成迁移脚本
flask db upgrade   # 应用迁移到数据库

最佳实践

项目结构

使用应用工厂模式：

# app/__init__.py
def create_app(config_name='default'):
    app = Flask(__name__)
    
    # 加载配置
    app.config.from_object(config[config_name])
    
    # 初始化扩展
    db.init_app(app)
    login_manager.init_app(app)
    
    # 注册蓝图
    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)
    
    from .auth import auth as auth_blueprint
    app.register_blueprint(auth_blueprint, url_prefix='/auth')
    
    return app

配置管理

使用类管理不同环境的配置：

# config.py
import os

class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess-string'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

错误处理

集中式错误处理：

from flask import render_template

# 在应用工厂中注册错误处理器
def create_app():
    app = Flask(__name__)
    
    # ... 其他设置 ...
    
    @app.errorhandler(404)
    def page_not_found(e):
        return render_template('errors/404.html'), 404

    @app.errorhandler(500)
    def internal_server_error(e):
        return render_template('errors/500.html'), 500
    
    return app

总结

Flask 是一个非常灵活且强大的 Python Web 框架，适合从小型应用到大型项目的各种需求。本文介绍了 Flask 的核心概念和实践技巧，包括路由系统、模板引擎、数据库集成、身份验证、RESTful API 开发以及部署等方面。

通过掌握这些知识和技能，你可以使用 Flask 构建出高效、安全、可维护的 Web 应用。随着项目的不断发展，良好的项目结构和编程实践将有助于保持代码的可读性和可维护性。