Skip to content

Flask框架

Flask是一个轻量级的Python Web框架,被设计为简单而易于扩展。它被称为"微框架",因为它不依赖特定的工具或库,开发者可以自由选择组件来构建应用。Flask由Armin Ronacher创建,最初是愚人节的一个玩笑项目,后来发展成为一个成熟的框架。

Flask核心概念

WSGI应用

Flask应用是WSGI(Web Server Gateway Interface)应用的实例:

python
from flask import Flask

app = Flask(__name__)

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

if __name__ == '__main__':
    app.run(debug=True)

路由系统

Flask使用装饰器定义URL路由:

python
@app.route('/users/<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方法限制
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 处理登录
        return 'Logged in'
    else:
        # 显示登录表单
        return 'Login form'

请求和响应

Flask提供了全局的request对象来访问请求数据:

python
from flask import request

@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return 'No file part'
    
    file = request.files['file']
    if file.filename == '':
        return 'No selected file'
    
    # 处理文件上传
    file.save('/path/to/uploads/' + file.filename)
    return 'File uploaded successfully'

响应可以是字符串、元组或Response对象:

python
from flask import make_response, jsonify

@app.route('/api/data')
def get_data():
    data = {'name': 'John', 'age': 30}
    return jsonify(data)

@app.route('/cookie')
def set_cookie():
    resp = make_response('Cookie set!')
    resp.set_cookie('username', 'flask-user')
    return resp

模板渲染

Flask使用Jinja2作为模板引擎:

python
from flask import render_template

@app.route('/hello/<name>')
def hello(name):
    return render_template('hello.html', name=name)
html
<!-- templates/hello.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Hello</title>
</head>
<body>
    <h1>Hello, {{ name }}!</h1>
    {% if name == 'admin' %}
        <p>Welcome, administrator!</p>
    {% else %}
        <p>Welcome, user!</p>
    {% endif %}
</body>
</html>

会话和Cookie

Flask提供了session对象用于跨请求存储数据:

python
from flask import session, redirect, url_for

app.secret_key = 'your_secret_key'  # 用于加密会话数据

@app.route('/login', methods=['POST'])
def login():
    session['username'] = request.form['username']
    return redirect(url_for('profile'))

@app.route('/profile')
def profile():
    if 'username' in session:
        return f'Logged in as {session["username"]}'
    return redirect(url_for('login'))

@app.route('/logout')
def logout():
    session.pop('username', None)
    return redirect(url_for('index'))

Flask应用结构

简单应用结构

适用于小型应用:

my_flask_app/
├── app.py                  # 应用入口
├── templates/              # 模板文件夹
│   └── index.html
├── static/                 # 静态文件夹
│   ├── css/
│   ├── js/
│   └── images/
├── requirements.txt        # 项目依赖
└── venv/                   # 虚拟环境

模块化应用结构

适用于中型应用:

my_flask_app/
├── app/
│   ├── __init__.py         # 应用初始化
│   ├── views.py            # 视图函数
│   ├── models.py           # 数据模型
│   ├── forms.py            # 表单定义
│   ├── templates/          # 模板文件夹
│   └── static/             # 静态文件夹
├── config.py               # 配置文件
├── run.py                  # 运行脚本
├── requirements.txt        # 项目依赖
└── venv/                   # 虚拟环境

蓝图结构

适用于大型应用:

my_flask_app/
├── app/
│   ├── __init__.py         # 应用初始化
│   ├── extensions.py       # 扩展实例化
│   ├── config.py           # 配置文件
│   ├── models/             # 数据模型
│   ├── api/                # API蓝图
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   └── ...
│   ├── admin/              # 管理后台蓝图
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   └── ...
│   ├── auth/               # 认证蓝图
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   └── ...
│   ├── templates/          # 模板文件夹
│   └── static/             # 静态文件夹
├── migrations/             # 数据库迁移
├── tests/                  # 测试代码
├── manage.py               # 管理脚本
├── requirements.txt        # 项目依赖
└── venv/                   # 虚拟环境

Flask蓝图

蓝图用于组织相关的视图、模板和静态文件:

python
# app/auth/__init__.py
from flask import Blueprint

auth_bp = Blueprint('auth', __name__, template_folder='templates')

from . import routes
python
# app/auth/routes.py
from flask import render_template, redirect, url_for
from . 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')
python
# app/__init__.py
from flask import Flask

def create_app(config_name):
    app = Flask(__name__)
    # 加载配置
    
    # 注册蓝图
    from .auth import auth_bp
    app.register_blueprint(auth_bp, url_prefix='/auth')
    
    # 其他蓝图注册
    
    return app

Flask扩展

Flask的生态系统包含许多扩展,用于扩展框架功能:

数据库集成

Flask-SQLAlchemy

提供SQLAlchemy的支持:

python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.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)
    
    def __repr__(self):
        return f'<User {self.username}>'

Flask-Migrate

用于数据库迁移:

python
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)

# 使用命令行
# flask db init      # 创建迁移仓库
# flask db migrate   # 创建迁移脚本
# flask db upgrade   # 应用迁移

表单处理

Flask-WTF

提供WTForms集成:

python
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('Login')
python
@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)

认证

Flask-Login

处理用户会话管理:

python
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user

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

class User(db.Model, UserMixin):
    # 模型定义...
    pass

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

@app.route('/login', methods=['POST'])
def login():
    # 验证用户...
    login_user(user, remember=True)
    return redirect(url_for('index'))

@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')

RESTful API

Flask-RESTful

用于构建RESTful API:

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

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

todos = {}

class TodoResource(Resource):
    def get(self, todo_id):
        return {todo_id: todos.get(todo_id, "")}
    
    def put(self, todo_id):
        parser = reqparse.RequestParser()
        parser.add_argument('task', type=str, required=True)
        args = parser.parse_args()
        todos[todo_id] = args['task']
        return {todo_id: todos[todo_id]}
    
    def delete(self, todo_id):
        if todo_id in todos:
            del todos[todo_id]
        return '', 204

api.add_resource(TodoResource, '/todos/<int:todo_id>')

Flask-Marshmallow

用于序列化和反序列化:

python
from flask_marshmallow import Marshmallow

app = Flask(__name__)
ma = Marshmallow(app)

class UserSchema(ma.Schema):
    class Meta:
        fields = ('id', 'username', 'email')

user_schema = UserSchema()
users_schema = UserSchema(many=True)

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return user_schema.jsonify(user)

@app.route('/api/users')
def get_users():
    users = User.query.all()
    return users_schema.jsonify(users)

测试

Flask Testing Tools

Flask自带测试客户端:

python
import unittest
from app import create_app, db

class TestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        self.client = self.app.test_client()
    
    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.assertTrue('Welcome' in response.get_data(as_text=True))

Flask应用实例

简单API服务

python
from flask import Flask, jsonify, request, abort
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tasks.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    description = db.Column(db.Text)
    done = db.Column(db.Boolean, default=False)
    
    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'description': self.description,
            'done': self.done
        }

@app.route('/api/tasks', methods=['GET'])
def get_tasks():
    tasks = Task.query.all()
    return jsonify([task.to_dict() for task in tasks])

@app.route('/api/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id):
    task = Task.query.get_or_404(task_id)
    return jsonify(task.to_dict())

@app.route('/api/tasks', methods=['POST'])
def create_task():
    if not request.json or 'title' not in request.json:
        abort(400)
    task = Task(
        title=request.json['title'],
        description=request.json.get('description', ''),
        done=request.json.get('done', False)
    )
    db.session.add(task)
    db.session.commit()
    return jsonify(task.to_dict()), 201

@app.route('/api/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
    task = Task.query.get_or_404(task_id)
    if not request.json:
        abort(400)
    task.title = request.json.get('title', task.title)
    task.description = request.json.get('description', task.description)
    task.done = request.json.get('done', task.done)
    db.session.commit()
    return jsonify(task.to_dict())

@app.route('/api/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    task = Task.query.get_or_404(task_id)
    db.session.delete(task)
    db.session.commit()
    return '', 204

if __name__ == '__main__':
    db.create_all()
    app.run(debug=True)

基于蓝图的Web应用

python
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from config import config

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

def create_app(config_name):
    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
python
# app/auth/routes.py
from flask import render_template, redirect, url_for, flash
from flask_login import login_user, logout_user, login_required
from . import auth
from .forms import LoginForm, RegistrationForm
from .. import db
from ..models import User

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user is not None and user.verify_password(form.password.data):
            login_user(user, form.remember_me.data)
            return redirect(url_for('main.index'))
        flash('Invalid email or password.')
    return render_template('auth/login.html', form=form)

@auth.route('/logout')
@login_required
def logout():
    logout_user()
    flash('You have been logged out.')
    return redirect(url_for('main.index'))

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegistrationForm()
    if form.validate_on_submit():
        user = User(email=form.email.data, username=form.username.data, password=form.password.data)
        db.session.add(user)
        db.session.commit()
        flash('You can now login.')
        return redirect(url_for('auth.login'))
    return render_template('auth/register.html', form=form)

Flask应用部署

使用Gunicorn部署

bash
# 安装Gunicorn
pip install gunicorn

# 运行
gunicorn -w 4 -b 127.0.0.1:8000 wsgi:app
python
# wsgi.py
from app import create_app

app = create_app('production')

if __name__ == '__main__':
    app.run()

使用Docker部署

dockerfile
# Dockerfile
FROM python:3.9-slim

WORKDIR /app

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

COPY .. .

ENV FLASK_APP=wsgi.py
ENV FLASK_ENV=production

EXPOSE 5000

CMD ["gunicorn", "--bind", "0.0.0.0:5000", "wsgi:app"]
yaml
# docker-compose.yml
version: '3'

services:
  web:
    build: .
    ports:
      - "5000:5000"
    environment:
      - FLASK_APP=wsgi.py
      - FLASK_ENV=production
      - DATABASE_URL=postgresql://user:password@db:5432/dbname
    depends_on:
      - db
  
  db:
    image: postgres:13
    volumes:
      - postgres_data:/var/lib/postgresql/data
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=dbname

volumes:
  postgres_data:

Flask最佳实践

应用工厂模式

使用工厂函数创建应用实例,便于测试和配置管理:

python
def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    
    # 初始化扩展
    # 注册蓝图
    
    return app

使用环境变量

通过环境变量管理敏感配置:

python
import os

class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard to guess string'
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'

错误处理

自定义错误页面和API错误响应:

python
@app.errorhandler(404)
def page_not_found(e):
    if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
        return jsonify(error='not found'), 404
    return render_template('404.html'), 404

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

日志配置

设置合适的日志级别和处理器:

python
import logging
from logging.handlers import RotatingFileHandler
import os

def configure_logging(app):
    if not app.debug and not app.testing:
        if not os.path.exists('logs'):
            os.mkdir('logs')
        file_handler = RotatingFileHandler('logs/app.log', maxBytes=10240, backupCount=10)
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
        ))
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)
        app.logger.setLevel(logging.INFO)
        app.logger.info('Application startup')

安全最佳实践

  1. 设置密钥:使用强随机密钥
  2. CSRF保护:使用Flask-WTF提供的CSRF保护
  3. 安全Cookie:设置安全Cookie标志
  4. 输入验证:验证所有用户输入
  5. 参数化SQL查询:使用ORM或参数化查询防止SQL注入

Flask与其他框架对比

Flask vs Django

特性FlaskDjango
规模微框架全栈框架
哲学简单、灵活包含电池、约定优于配置
学习曲线中到高
数据库支持通过扩展内置ORM
管理后台通过扩展内置
用户认证通过扩展内置
适用场景小型到中型应用、API、原型大型应用、内容管理、完整Web应用

Flask vs FastAPI

特性FlaskFastAPI
性能良好优秀(基于Starlette)
异步支持有限原生支持
API文档通过扩展内置(Swagger/ReDoc)
类型提示可选核心特性
数据验证通过扩展内置(基于Pydantic)
社区与生态成熟、庞大增长迅速

总结

Flask是一个灵活、简单且强大的Python Web框架,其微框架性质使开发者能够只使用所需的组件,而不必应对过多的约束。Flask的优势在于:

  1. 简单易学:最小化的核心,易于入门
  2. 灵活性:通过扩展系统可以根据需要添加功能
  3. 可定制性:几乎所有组件都可以替换或定制
  4. 轻量级:核心功能简洁,易于理解

这些特性使Flask成为原型开发、小型应用和API服务的理想选择。对于经验丰富的开发者,Flask提供了构建复杂应用所需的所有工具,同时保持了代码的简洁和可维护性。

用知识点燃技术的火山