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')
安全最佳实践
- 设置密钥:使用强随机密钥
- CSRF保护:使用Flask-WTF提供的CSRF保护
- 安全Cookie:设置安全Cookie标志
- 输入验证:验证所有用户输入
- 参数化SQL查询:使用ORM或参数化查询防止SQL注入
Flask与其他框架对比
Flask vs Django
特性 | Flask | Django |
---|---|---|
规模 | 微框架 | 全栈框架 |
哲学 | 简单、灵活 | 包含电池、约定优于配置 |
学习曲线 | 低 | 中到高 |
数据库支持 | 通过扩展 | 内置ORM |
管理后台 | 通过扩展 | 内置 |
用户认证 | 通过扩展 | 内置 |
适用场景 | 小型到中型应用、API、原型 | 大型应用、内容管理、完整Web应用 |
Flask vs FastAPI
特性 | Flask | FastAPI |
---|---|---|
性能 | 良好 | 优秀(基于Starlette) |
异步支持 | 有限 | 原生支持 |
API文档 | 通过扩展 | 内置(Swagger/ReDoc) |
类型提示 | 可选 | 核心特性 |
数据验证 | 通过扩展 | 内置(基于Pydantic) |
社区与生态 | 成熟、庞大 | 增长迅速 |
总结
Flask是一个灵活、简单且强大的Python Web框架,其微框架性质使开发者能够只使用所需的组件,而不必应对过多的约束。Flask的优势在于:
- 简单易学:最小化的核心,易于入门
- 灵活性:通过扩展系统可以根据需要添加功能
- 可定制性:几乎所有组件都可以替换或定制
- 轻量级:核心功能简洁,易于理解
这些特性使Flask成为原型开发、小型应用和API服务的理想选择。对于经验丰富的开发者,Flask提供了构建复杂应用所需的所有工具,同时保持了代码的简洁和可维护性。