跳到主要内容
Python Web 开发:Flask 框架从入门到实战指南 | 极客日志
Python
Python Web 开发:Flask 框架从入门到实战指南 Flask 是 Python 轻量级 Web 框架,以灵活性和扩展性著称。本文涵盖安装配置、路由系统、模板引擎 Jinja2、表单处理、数据库集成(SQLAlchemy)、用户认证(Flask-Login)、REST API 开发、蓝图架构及生产环境部署(Gunicorn/Docker)。通过实战代码示例,解析项目结构设计与最佳实践,帮助开发者快速构建安全高效的 Web 应用。
292440837 发布于 2026/4/7 更新于 2026/4/25 1 浏览Flask 是 Python Web 开发领域最受欢迎的微框架之一,以其轻量、灵活和易于扩展的特性赢得了众多开发者的青睐。无论是构建简单的 API 服务,还是开发功能完备的 Web 应用,Flask 都能提供优雅而高效的解决方案。
无论你是 Web 开发新手,还是想从其他框架迁移到 Flask,这篇指南都将为你提供系统化的学习路径,帮助你构建专业、高效且安全的 Python Web 应用。
Flask 简介
什么是 Flask
Flask 是一个轻量级的 Python Web 应用框架,由 Armin Ronacher 设计开发,基于 Werkzeug WSGI 工具包和 Jinja2 模板引擎。Flask 被称为'微框架',因为它保持核心简单但可扩展,不强制依赖特定的库或工具,给予开发者极大的灵活性和控制力。
Flask 的主要特点包括:
轻量且高效 :核心简洁,启动迅速,资源占用低
灵活性 :不强制特定项目结构或组件选择
易于学习 :API 设计直观,学习曲线平缓
可扩展性 :通过丰富的扩展生态系统增强功能
强大的路由系统 :支持 URL 变量和 HTTP 方法
内置开发服务器 :便于本地测试和开发
RESTful 支持 :轻松构建符合 REST 规范的 API
Flask vs. Django
Flask 和 Django 是 Python Web 开发中最流行的两个框架,它们各有优势:
特性 Flask Django 架构理念 微框架,灵活定制 全能框架,内置齐全 学习曲线 较低,容易上手 较高,概念较多 项目规模 适合小到中型项目 适合中到大型项目 自由度 高,可自由选择组件 低,遵循"Django 方式" 数据库支持 通过扩展支持 ORM 内置 管理后台 需要自行实现或使用扩展 内置强大的管理后台
安装 Flask
使用 pip 安装 Flask 非常简单:
pip install flask
建议在虚拟环境中安装 Flask,以避免依赖冲突:
python -m venv venv
venv\Scripts\activate
source venv/bin/activate
pip install flask
验证安装:
python -c "import flask; print(flask.__version__)"
Flask 基础知识
第一个 Flask 应用
创建一个最简单的 Flask 应用只需几行代码:
from flask import 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/ 即可看到"Hello, World!"消息。
应用实例 Flask 应用的核心是 Flask 类的实例,通常命名为 app:
参数 __name__ 是 Python 的特殊变量,它会传递当前模块的名称给 Flask。这有助于 Flask 找到资源文件的位置。
路由系统 路由是将 URL 映射到视图函数的机制。Flask 使用装饰器来定义路由:
@app.route('/user/<username>' )
def show_user_profile (username ):
return f'User {username} '
URL 变量 Flask 支持在 URL 中包含变量,类型可以是:
字符串(默认):<username>
整数:<int:post_id>
浮点数:<float:score>
路径:<path:subpath>
UUID:<uuid:id>
@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 '处理登录'
else :
return '显示登录表单'
视图函数 视图函数是处理请求并返回响应的 Python 函数。视图函数可以返回:
字符串:直接显示为 HTML
HTML 模板渲染结果
JSON 响应
重定向
自定义响应对象
from flask import render_template, jsonify, redirect, url_for
@app.route('/template' )
def template_example ():
return render_template('example.html' , name='Flask' )
@app.route('/api/data' )
def api_data ():
return jsonify({"name" : "Flask" , "type" : "framework" })
@app.route('/redirect' )
def redirect_example ():
return redirect(url_for('hello_world' ))
请求对象 Flask 通过 request 对象提供对客户端请求数据的访问:
from flask import request
@app.route('/submit' , methods=['POST' ] )
def submit ():
username = request.form.get('username' )
page = request.args.get('page' , 1 , type =int )
data = request.get_json()
file = request.files.get('upload' )
return f'Received: {username} '
响应对象 视图函数可以返回一个元组来设置响应的状态码和头信息:
@app.route('/response' )
def custom_response ():
return 'Custom response' , 201 , {'X-Custom-Header' : 'value' }
也可以使用 make_response 函数创建自定义响应:
from flask import make_response
@app.route('/cookie' )
def set_cookie ():
resp = make_response('Cookie 设置成功' )
resp.set_cookie('username' , 'flask_user' )
return resp
模板系统 Flask 使用 Jinja2 作为默认的模板引擎,它功能强大且易于使用。
Jinja2 模板基础 Jinja2 模板是包含静态内容和动态内容占位符的文件。默认情况下,Flask 在应用的 templates 目录中查找模板。
一个基本的 HTML 模板示例(templates/index.html):
<!DOCTYPE html >
<html >
<head >
<title > {{ title }}</title >
</head >
<body >
<h1 > Hello, {{ name }}!</h1 >
{% if messages %}
<ul >
{% for message in messages %}
<li > {{ message }}</li >
{% endfor %}
</ul >
{% else %}
<p > No messages.</p >
{% endif %}
</body >
</html >
@app.route('/' )
def index ():
return render_template('index.html' , title='Flask Template' , name='User' , messages=['Message 1' , 'Message 2' ])
模板语法
变量 :{{ variable }}
控制结构 :{% if condition %} ... {% endif %}
注释 :{# This is a comment #}
变量与过滤器 {{ name|capitalize }}
{{ text|truncate(100) }}
{{ data|tojson }}
safe: 标记内容为安全,不进行转义
escape: HTML 转义
capitalize: 首字母大写
lower/upper: 转换大小写
trim: 去除首尾空白
striptags: 移除 HTML 标签
default: 提供默认值
控制结构 {% if user.is_authenticated %}
<a href="{{ url_for('logout') }}">Logout</a>
{% else %}
<a href="{{ url_for('login') }}">Login</a>
{% endif %}
<ul>
{% for item in items %}
<li>{{ loop.index }} - {{ item.name }}</li>
{% else %}
<li>No items found.</li>
{% endfor %}
</ul>
模板继承 Jinja2 支持模板继承,这是一种强大的组织模板的方式。
<!DOCTYPE html >
<html >
<head >
<title > {% block title %}Default Title{% endblock %}</title >
<link rel ="stylesheet" href ="{{ url_for('static', filename='style.css') }}" >
{% block styles %}{% endblock %}
</head >
<body >
<header >
<nav > {% block nav %}{% endblock %}</nav >
</header >
<main >
{% block content %}{% endblock %}
</main >
<footer >
{% block footer %}© 2023 Flask 应用{% endblock %}
</footer >
{% block scripts %}{% endblock %}
</body >
</html >
{% extends "base.html" %}
{% block title %}页面标题{% endblock %}
{% block content %}
<h1>页面内容</h1>
<p>这是页面的具体内容。</p>
{% endblock %}
静态文件 Flask 自动为静态文件添加路由。默认情况下,静态文件应放在应用的 static 目录中。
<link rel ="stylesheet" href ="{{ url_for('static', filename='css/style.css') }}" >
<img src ="{{ url_for('static', filename='images/logo.png') }}" >
<script src ="{{ url_for('static', filename='js/script.js') }}" > </script >
URL 生成 使用 url_for() 函数动态生成 URL,避免硬编码:
<a href ="{{ url_for('index') }}" > 首页</a >
<a href ="{{ url_for('user_profile', username='john') }}" > 用户资料</a >
<a href ="{{ url_for('static', filename='style.css') }}" > 样式表</a >
表单处理 Web 应用几乎都需要处理用户输入的表单数据。Flask 提供了多种方式处理表单提交。
基本表单处理 最简单的表单处理方式是直接使用 Flask 的 request 对象:
from flask import request, redirect, url_for, render_template
@app.route('/login' , methods=['GET' , 'POST' ] )
def login ():
if request.method == 'POST' :
username = request.form.get('username' )
password = request.form.get('password' )
if username == 'admin' and password == 'secret' :
return redirect(url_for('dashboard' ))
else :
error = '无效的用户名或密码'
return render_template('login.html' , error=error)
return render_template('login.html' )
<!DOCTYPE html >
<html >
<head > <title > 登录</title > </head >
<body >
<h1 > 登录</h1 >
{% if error %}
<p style ="color: red;" > {{ error }}</p >
{% endif %}
<form method ="post" >
<div >
<label > 用户名:</label >
<input type ="text" name ="username" required >
</div >
<div >
<label > 密码:</label >
<input type ="password" name ="password" required >
</div >
<button type ="submit" > 登录</button >
</form >
</body >
</html >
使用 Flask-WTF 扩展 对于复杂表单,推荐使用 Flask-WTF 扩展,它结合了 WTForms 库,提供了表单验证、CSRF 保护等功能。
app = Flask(__name__)
app.config['SECRET_KEY' ] = 'your-secret-key'
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length
class LoginForm (FlaskForm ):
email = StringField('邮箱' , validators=[DataRequired(), Email()])
password = PasswordField('密码' , validators=[DataRequired(), Length(min =6 )])
submit = SubmitField('登录' )
@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('dashboard' ))
return render_template('login_wtf.html' , form=form)
带有 WTForms 的模板(login_wtf.html):
<!DOCTYPE html >
<html >
<head > <title > 登录</title > </head >
<body >
<h1 > 登录</h1 >
<form method ="post" >
{{ form.hidden_tag() }}
<div >
{{ form.email.label }}
{{ form.email }}
{% if form.email.errors %}
<ul >
{% for error in form.email.errors %}
<li > {{ error }}</li >
{% endfor %}
</ul >
{% endif %}
</div >
<div >
{{ form.password.label }}
{{ form.password }}
{% if form.password.errors %}
<ul >
{% for error in form.password.errors %}
<li > {{ error }}</li >
{% endfor %}
</ul >
{% endif %}
</div >
{{ form.submit }}
</form >
</body >
</html >
文件上传 处理文件上传需要在表单中添加 enctype="multipart/form-data" 属性:
<form method ="post" enctype ="multipart/form-data" >
<input type ="file" name ="file" >
<button type ="submit" > 上传</button >
</form >
from werkzeug.utils import secure_filename
import os
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 '没有文件部分'
file = request.files['file' ]
if file.filename == '' :
return '未选择文件'
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>
<title>上传文件</title>
<h1>上传文件</h1>
<form method="post" enctype="multipart/form-data">
<input type="file" name="file">
<input type="submit" value="上传">
</form>'''
表单验证
DataRequired:字段不能为空
Email:必须是有效的电子邮件地址
Length:字符串长度限制
NumberRange:数值范围限制
EqualTo:字段必须与另一个字段值相等(如密码确认)
URL:必须是有效的 URL
Regexp:必须匹配正则表达式
from wtforms import ValidationError
class RegistrationForm (FlaskForm ):
username = StringField('用户名' , validators=[DataRequired(), Length(min =4 , max =20 )])
email = StringField('邮箱' , validators=[DataRequired(), Email()])
password = PasswordField('密码' , validators=[DataRequired(), Length(min =6 )])
confirm_password = PasswordField('确认密码' , validators=[DataRequired(), EqualTo('password' )])
submit = SubmitField('注册' )
def validate_username (self, username ):
if username.data == 'admin' :
raise ValidationError('该用户名已被使用,请选择其他用户名。' )
数据库集成 Flask 本身不包含数据库抽象层,但可以与各种数据库解决方案集成。最常用的是 SQLAlchemy ORM 通过 Flask-SQLAlchemy 扩展。
Flask-SQLAlchemy 基础 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)
SQLite: sqlite:///site.db
MySQL: mysql://username:password@localhost/db_name
PostgreSQL: postgresql://username:password@localhost/db_name
定义模型 使用 SQLAlchemy 定义数据库模型(表):
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 )
password = db.Column(db.String(60 ), nullable=False )
posts = db.relationship('Post' , backref='author' , lazy=True )
def __repr__ (self ):
return f"User('{self.username} ', '{self.email} ')"
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 )
date_posted = db.Column(db.DateTime, nullable=False , default=datetime.utcnow)
user_id = db.Column(db.Integer, db.ForeignKey('user.id' ), nullable=False )
def __repr__ (self ):
return f"Post('{self.title} ', '{self.date_posted} ')"
创建和迁移数据库
from app import db
db.create_all()
对于更复杂的迁移,可以使用 Flask-Migrate 扩展(基于 Alembic):
pip install flask-migrate
from flask_migrate import Migrate
migrate = Migrate(app, db)
flask db init
flask db migrate
flask db upgrade
基本 CRUD 操作
创建记录 @app.route('/add_user' , methods=['POST' ] )
def add_user ():
username = request.form.get('username' )
email = request.form.get('email' )
password = request.form.get('password' )
user = User(username=username, email=email, password=password)
db.session.add(user)
db.session.commit()
return f'用户 {username} 已添加'
查询记录 @app.route('/users' )
def get_users ():
users = User.query.all ()
return render_template('users.html' , users=users)
@app.route('/user/<int:user_id>' )
def get_user (user_id ):
user = User.query.get_or_404(user_id)
return render_template('user.html' , user=user)
User.query.all ()
User.query.get(1 )
User.query.get_or_404(1 )
User.query.filter_by(username='john' ).first()
User.query.filter (User.email.endswith('@example.com' )).all ()
User.query.order_by(User.username).all ()
User.query.limit(10 ).all ()
users = User.query.paginate(page=2 , per_page=20 )
for user in users.items:
print (user.username)
更新记录 @app.route('/update_user/<int:user_id>' , methods=['POST' ] )
def update_user (user_id ):
user = User.query.get_or_404(user_id)
user.username = request.form.get('username' , user.username)
user.email = request.form.get('email' , user.email)
db.session.commit()
return f'用户 {user.username} 已更新'
删除记录 @app.route('/delete_user/<int:user_id>' , methods=['POST' ] )
def delete_user (user_id ):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return f'用户 {user.username} 已删除'
使用其他数据库
SQLite app.config['SQLALCHEMY_DATABASE_URI' ] = 'sqlite:///site.db'
MySQL app.config['SQLALCHEMY_DATABASE_URI' ] = 'mysql://username:password@localhost/db_name'
PostgreSQL pip install psycopg2-binary
app.config['SQLALCHEMY_DATABASE_URI' ] = 'postgresql://username:password@localhost/db_name'
MongoDB (NoSQL) 对于 MongoDB 等 NoSQL 数据库,可以使用 Flask-PyMongo 或者 Flask-MongoEngine 扩展。
pip install flask-mongoengine
from flask_mongoengine import MongoEngine
app = Flask(__name__)
app.config['MONGODB_SETTINGS' ] = {
'db' : 'your_database' ,
'host' : 'localhost' ,
'port' : 27017
}
db = MongoEngine(app)
class User (db.Document):
email = db.StringField(required=True , unique=True )
username = db.StringField(required=True , unique=True )
password = db.StringField(required=True )
用户认证与授权 大多数 Web 应用需要用户认证与授权功能。Flask 通过扩展提供了丰富的认证解决方案。
Flask-Login 扩展 Flask-Login 提供了用户 session 管理、登录、登出等功能。
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
app = Flask(__name__)
app.config['SECRET_KEY' ] = 'your-secret-key'
login_manager = LoginManager(app)
login_manager.login_view = 'login'
login_manager.login_message = '请先登录再访问此页面。'
@login_manager.user_loader
def load_user (user_id ):
return User.query.get(int (user_id))
class User (db.Model, UserMixin):
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 )
password = db.Column(db.String(60 ), nullable=False )
@app.route('/login' , methods=['GET' , 'POST' ] )
def login ():
if current_user.is_authenticated:
return redirect(url_for('index' ))
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user and check_password(form.password.data, user.password):
login_user(user, remember=form.remember.data)
next_page = request.args.get('next' )
if next_page:
return redirect(next_page)
else :
return redirect(url_for('index' ))
else :
flash('登录失败。请检查邮箱和密码。' )
return render_template('login.html' , form=form)
@app.route('/logout' )
def logout ():
logout_user()
return redirect(url_for('login' ))
@app.route('/profile' )
@login_required
def profile ():
return render_template('profile.html' )
{% if current_user.is_authenticated %}
<p>当前用户:{{ current_user.username }}</p>
<a href="{{ url_for('logout') }}">退出登录</a>
{% else %}
<a href="{{ url_for('login') }}">登录</a>
<a href="{{ url_for('register') }}">注册</a>
{% endif %}
密码哈希 不应该明文存储密码,应使用加密哈希。Flask-Bcrypt 是一个优秀的密码哈希扩展:
from flask_bcrypt import Bcrypt
bcrypt = Bcrypt(app)
hashed_password = bcrypt.generate_password_hash('password' ).decode('utf-8' )
valid = bcrypt.check_password_hash(hashed_password, 'password' )
@app.route('/register' , methods=['GET' , 'POST' ] )
def register ():
form = RegistrationForm()
if form.validate_on_submit():
hashed_password = bcrypt.generate_password_hash(form.password.data).decode('utf-8' )
user = User(username=form.username.data, email=form.email.data, password=hashed_password)
db.session.add(user)
db.session.commit()
flash(f'账户已创建,现在可以登录了!' )
return redirect(url_for('login' ))
return render_template('register.html' , form=form)
基于角色的访问控制 对于更复杂的权限控制,可以使用 Flask-Principal 或实现自定义角色系统:
class User (db.Model, UserMixin):
role = db.Column(db.String(20 ), nullable=False , default='user' )
def is_admin (self ):
return self .role == 'admin'
from functools import wraps
from flask import abort
def admin_required (f ):
@wraps(f )
def decorated_function (*args, **kwargs ):
if not current_user.is_authenticated or not current_user.is_admin():
abort(403 )
return f(*args, **kwargs)
return decorated_function
@app.route('/admin' )
@login_required
@admin_required
def admin_panel ():
return render_template('admin/index.html' )
Flask-Security 扩展 对于更全面的安全解决方案,Flask-Security 集成了多种安全扩展:
pip install flask-security-too
用户认证
角色管理
密码哈希
基本 HTTP 认证
令牌认证
用户注册
密码重置
邮件确认
from flask_security import Security, SQLAlchemyUserDatastore
class Role (db.Model, RoleMixin):
id = db.Column(db.Integer, primary_key=True )
name = db.Column(db.String(80 ), unique=True )
description = db.Column(db.String(255 ))
class User (db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True )
email = db.Column(db.String(255 ), unique=True )
password = db.Column(db.String(255 ))
active = db.Column(db.Boolean())
confirmed_at = db.Column(db.DateTime())
roles = db.relationship('Role' , secondary=roles_users, backref=db.backref('users' , lazy='dynamic' ))
user_datastore = SQLAlchemyUserDatastore(db, User, Role)
security = Security(app, user_datastore)
REST API 开发 Flask 非常适合构建 RESTful API。本节将介绍如何使用 Flask 开发 API 服务。
基本 API 端点 最简单的 API 可以直接使用 Flask 的视图函数和 jsonify 函数:
from flask import jsonify
@app.route('/api/users' )
def get_users ():
users = User.query.all ()
user_list = []
for user in users:
user_data = {
'id' : user.id ,
'username' : user.username,
'email' : user.email
}
user_list.append(user_data)
return jsonify({'users' : user_list})
@app.route('/api/user/<int:user_id>' )
def get_user (user_id ):
user = User.query.get_or_404(user_id)
user_data = {
'id' : user.id ,
'username' : user.username,
'email' : user.email
}
return jsonify(user_data)
@app.route('/api/users' , methods=['POST' ] )
def create_user ():
if not request.json or not 'username' in request.json:
abort(400 )
user = User(
username=request.json['username' ],
email=request.json.get('email' , '' )
)
db.session.add(user)
db.session.commit()
return jsonify({'user' : {
'id' : user.id ,
'username' : user.username,
'email' : user.email
}}), 201
@app.route('/api/user/<int:user_id>' , methods=['PUT' ] )
def update_user (user_id ):
user = User.query.get_or_404(user_id)
if not request.json:
abort(400 )
user.username = request.json.get('username' , user.username)
user.email = request.json.get('email' , user.email)
db.session.commit()
return jsonify({'user' : {
'id' : user.id ,
'username' : user.username,
'email' : user.email
}})
使用 Flask-RESTful 扩展 Flask-RESTful 提供了更结构化的 API 开发方式:
pip install flask-restful
from flask import Flask
from flask_restful import Api, Resource, reqparse, fields, marshal_with
app = Flask(__name__)
api = Api(app)
user_fields = {
'id' : fields.Integer,
'username' : fields.String,
'email' : fields.String,
'uri' : fields.Url('user' )
}
user_parser = reqparse.RequestParser()
user_parser.add_argument('username' , type =str , required=True , help ='用户名不能为空' )
user_parser.add_argument('email' , type =str , required=True , help ='邮箱不能为空' )
class UserResource (Resource ):
@marshal_with(user_fields )
def get (self, user_id ):
user = User.query.get_or_404(user_id)
return user
@marshal_with(user_fields )
def put (self, user_id ):
args = user_parser.parse_args()
user = User.query.get_or_404(user_id)
user.username = args['username' ]
user.email = args['email' ]
db.session.commit()
return user
def delete (self, user_id ):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '' , 204
class UserListResource (Resource ):
@marshal_with(user_fields )
def get (self ):
users = User.query.all ()
return users
@marshal_with(user_fields )
def post (self ):
args = user_parser.parse_args()
user = User(username=args['username' ], email=args['email' ])
db.session.add(user)
db.session.commit()
return user, 201
api.add_resource(UserListResource, '/api/users' )
api.add_resource(UserResource, '/api/user/<int:user_id>' , endpoint='user' )
API 认证
基本认证 from flask_httpauth import HTTPBasicAuth
auth = HTTPBasicAuth()
@auth.verify_password
def verify_password (username, password ):
user = User.query.filter_by(username=username).first()
if user and bcrypt.check_password_hash(user.password, password):
return user
return None
class ProtectedResource (Resource ):
@auth.login_required
def get (self ):
return {'message' : '只有认证用户才能看到' }
令牌认证 from flask_httpauth import HTTPTokenAuth
from itsdangerous import Serializer
auth = HTTPTokenAuth()
@auth.verify_token
def verify_token (token ):
user = User.verify_token(token)
if user:
return user
return None
class User (db.Model):
def generate_token (self, expiration=3600 ):
s = Serializer(app.config['SECRET_KEY' ], expires_in=expiration)
return s.dumps({'id' : self .id }).decode('utf-8' )
@staticmethod
def verify_token (token ):
s = Serializer(app.config['SECRET_KEY' ])
try :
data = s.loads(token)
except :
return None
return User.query.get(data['id' ])
API 文档生成 自动生成 API 文档可以使用 Flask-RESTPlus 或 Swagger-UI:
pip install flask-restplus
from flask import Flask
from flask_restplus import Api, Resource, fields
app = Flask(__name__)
api = Api(app, version='1.0' , title='用户 API' , description='用户管理 API 文档' )
ns = api.namespace('users' , description='用户操作' )
user_model = api.model('User' , {
'id' : fields.Integer(readonly=True , description='用户 ID' ),
'username' : fields.String(required=True , description='用户名' ),
'email' : fields.String(required=True , description='邮箱地址' )
})
@ns.route('/' )
class UserList (Resource ):
@ns.doc('列出所有用户' )
@ns.marshal_list_with(user_model )
def get (self ):
"""获取所有用户列表"""
return User.query.all ()
@ns.doc('创建用户' )
@ns.expect(user_model )
@ns.marshal_with(user_model, code=201 )
def post (self ):
"""创建新用户"""
user = User(username=api.payload['username' ], email=api.payload['email' ])
db.session.add(user)
db.session.commit()
return user, 201
@ns.route('/<int:id>' )
@ns.response(404 , '用户未找到' )
@ns.param('id' , '用户 ID' )
class UserAPI (Resource ):
@ns.doc('获取用户' )
@ns.marshal_with(user_model )
def get (self, id ):
"""获取指定 ID 的用户"""
return User.query.get_or_404(id )
访问 / 路径即可看到自动生成的 Swagger UI 文档。
API 版本控制 API 版本控制可以通过 URL 路径、请求头或查询参数实现:
URL 路径版本控制 @app.route('/api/v1/users' )
def get_users_v1 ():
pass
@app.route('/api/v2/users' )
def get_users_v2 ():
pass
使用蓝图实现版本控制 from flask import Blueprint
api_v1 = Blueprint('api_v1' , __name__, url_prefix='/api/v1' )
api_v2 = Blueprint('api_v2' , __name__, url_prefix='/api/v2' )
@api_v1.route('/users' )
def get_users_v1 ():
pass
@api_v2.route('/users' )
def get_users_v2 ():
pass
app.register_blueprint(api_v1)
app.register_blueprint(api_v2)
蓝图与应用结构 随着应用复杂度增加,将所有代码放在一个文件中变得不可维护。Flask 蓝图(Blueprint)提供了模块化应用的方式。
Flask 蓝图 蓝图是一种组织一组相关视图和其他代码的方式,可以被注册到应用,但不是一个完整的应用。
from flask import Blueprint
users_bp = Blueprint('users' , __name__, url_prefix='/users' )
@users_bp.route('/' )
def user_list ():
return 'User list'
@users_bp.route('/<int:user_id>' )
def user_detail (user_id ):
return f'User {user_id} detail'
from flask import Flask
from views.users import users_bp
app = Flask(__name__)
app.register_blueprint(users_bp)
蓝图高级用法 admin_bp = Blueprint('admin' , __name__, url_prefix='/admin' , template_folder='templates/admin' , static_folder='static/admin' )
@admin_bp.before_request
def restrict_to_admins ():
if not current_user.is_admin:
abort(403 )
from flask import Blueprint
main = Blueprint('main' , __name__)
admin = Blueprint('admin' , __name__, url_prefix='/admin' )
@main.route('/' )
def index ():
return 'Main index'
@admin.route('/' )
def admin_index ():
return 'Admin index'
super_bp = Blueprint('super' , __name__)
super_bp.register_blueprint(main)
super_bp.register_blueprint(admin)
app.register_blueprint(super_bp, url_prefix='/portal' )
应用工厂模式 应用工厂是一种设计模式,用于延迟创建应用实例,有助于测试和多实例部署:
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
from auth import auth as auth_blueprint
app.register_blueprint(main_blueprint)
app.register_blueprint(auth_blueprint, url_prefix='/auth' )
return app
import os
class Config :
SECRET_KEY = os.environ.get('SECRET_KEY' , 'hard-to-guess-string' )
SQLALCHEMY_TRACK_MODIFICATIONS = False
@staticmethod
def init_app (app ):
pass
class DevelopmentConfig (Config ):
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL' , 'sqlite:///dev.db' )
class TestingConfig (Config ):
TESTING = True
SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL' , 'sqlite:///:memory:' )
class ProductionConfig (Config ):
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL' , 'sqlite:///prod.db' )
config = {
'development' : DevelopmentConfig,
'testing' : TestingConfig,
'production' : ProductionConfig,
'default' : DevelopmentConfig
}
import os
from app import create_app, db
from app.models import User, Role
app = create_app(os.getenv('FLASK_CONFIG' , 'default' ))
@app.shell_context_processor
def make_shell_context ():
return dict (db=db, User=User, Role=Role)
推荐的项目结构 myapp/
├── app/
│ ├── __init__.py # 应用工厂
│ ├── models/ # 数据库模型
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── post.py
│ ├── views/ # 视图函数和蓝图
│ │ ├── __init__.py
│ │ ├── main.py
│ │ ├── auth.py
│ │ └── api.py
│ ├── forms/ # 表单类
│ │ ├── __init__.py
│ │ ├── auth.py
│ │ └── main.py
│ ├── static / # 静态文件
│ │ ├── css/
│ │ ├── js/
│ │ └── img/
│ ├── templates/ # HTML 模板
│ │ ├── base .html
│ │ ├── main/
│ │ ├── auth/
│ │ └── errors/
│ └── utils/ # 工具函数
│ ├── __init__.py
│ └── helpers.py
├── migrations/ # 数据库迁移
├── tests/ # 测试用例
│ ├── __init__.py
│ ├── test_user.py
│ └── test_api.py
├── venv/ # 虚拟环境
├── config.py # 配置文件
├── requirements.txt # 依赖包列表
├── run.py # 应用入口
└── README.md # 项目说明
包管理与依赖 pip freeze > requirements.txt
# requirements/base.txt
flask==2.0.1
flask-sqlalchemy==2.5.1
flask-login==0.5.0
# requirements/dev.txt
-r base.txt
pytest==6.2.5
coverage==6.0.2
# requirements/prod.txt
-r base.txt
gunicorn==20.1.0
pip install -r requirements/dev.txt
部署与维护 将 Flask 应用部署到生产环境需要考虑多方面因素,包括性能、安全和可靠性。
部署准备 from flask_caching import Cache
cache = Cache(app, config={'CACHE_TYPE' : 'SimpleCache' })
@app.route('/expensive-operation' )
@cache.cached(timeout=60 )
def expensive_operation ():
return result
import logging
logging.basicConfig(
filename='app.log' ,
level=logging.INFO,
format ='%(asctime)s %(levelname)s: %(message)s'
)
app.config['SECRET_KEY' ] = os.environ.get('SECRET_KEY' )
app.config['DATABASE_URI' ] = os.environ.get('DATABASE_URI' )
WSGI 服务器 Flask 内置的开发服务器不适用于生产环境,应使用生产级 WSGI 服务器:
Gunicorn gunicorn -w 4 -b 127.0.0.1:5000 wsgi:app
-w 4: 使用 4 个工作进程
-b 127.0.0.1:5000: 绑定地址和端口
wsgi:app: 应用入口点(wsgi.py 中的 app 对象)
uWSGI [uwsgi]
module = wsgi:app
master = true
processes = 4
socket = myapp.sock
chmod-socket = 660
vacuum = true
die-on-term = true
Web 服务器配置 为了处理静态文件、SSL 和请求分发,通常在 WSGI 服务器前配置 Nginx 或 Apache。
Nginx 配置 server {
listen 80;
server_name example.com;
location / {
proxy_pass http://127.0.0.1:5000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
location /static {
alias /path/to/your/app/static;
}
}
使用 HTTPS server {
listen 443 ssl;
server_name example.com;
ssl_certificate /path/to/cert.pem;
ssl_certificate_key /path/to/key.pem;
location / {
proxy_pass http://127.0.0.1:5000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
Docker 部署 FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV FLASK_APP=app
ENV FLASK_ENV=production
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "wsgi:app"]
docker build -t flask-app .
docker run -p 5000:5000 -d flask-app
使用 Docker Compose 管理多容器应用:
version: '3'
services:
web:
build: .
ports:
- "5000:5000"
environment:
- DATABASE_URI=postgresql://user:password@db:5432/app
depends_on:
- db
db:
image: postgres:13
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
- POSTGRES_DB=app
volumes:
- postgres_data:/var/lib/postgresql/data
volumes:
postgres_data:
部署平台
Heroku
PythonAnywhere import sys
path = '/home/yourusername/myapp'
if path not in sys.path:
sys.path.append(path)
from app import create_app
application = create_app('production' )
AWS Elastic Beanstalk 创建 .ebextensions/01_flask.config:
option_settings:
aws:elasticbeanstalk:container:python:WSGIPath: wsgi:app
性能监控与日志
ELK Stack - 日志收集与分析
Prometheus + Grafana - 指标监控:
pip install prometheus-flask-exporter
from prometheus_flask_exporter import PrometheusMetrics
metrics = PrometheusMetrics(app)
pip install sentry-sdk[flask]
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
sentry_sdk.init(
dsn="your-dsn-here" ,
integrations=[FlaskIntegration()]
)
自动化部署与 CI/CD GitHub Actions 配置示例(.github/workflows/deploy.yml):
name: Deploy
on:
push:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: '3.9'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pytest
pip install -r requirements.txt
- name: Test with pytest
run: |
pytest
deploy:
needs: test
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Deploy to Heroku
uses: akhileshns/[email protected]
with:
heroku_api_key: ${{secrets.HEROKU_API_KEY}}
heroku_app_name: "your-app-name"
heroku_email: "[email protected] "
总结 本文全面介绍了 Flask Web 框架的核心功能和开发最佳实践,覆盖了从基础入门到高级应用的各个方面。Flask 作为一个灵活而强大的微框架,其简约的设计理念和丰富的扩展生态系统使其成为 Python Web 开发的理想选择。
Flask 的优势
易于上手 :简单直观的 API 设计,学习曲线平缓
灵活性 :不强制特定项目结构或组件选择
可扩展性 :通过扩展实现各种高级功能
性能良好 :核心轻量且高效
活跃的社区 :丰富的文档和第三方库支持
学习路径建议
掌握基本路由和视图函数
学习模板系统和表单处理
探索数据库集成
实现用户认证与授权
熟悉应用结构和蓝图
学习 API 开发
了解高级主题和部署
进一步学习资源
结语 Flask 通过'微框架'的设计理念,提供了 Web 开发所需的核心功能,同时保持足够的灵活性让开发者根据自己的需求选择合适的工具和扩展。这种设计使 Flask 能够适应从简单 API 到复杂企业应用的各种开发场景。
通过本指南中介绍的概念和技术,你已经具备了使用 Flask 开发现代 Web 应用的基础知识。随着实践经验的积累,你将能够构建越来越复杂和高质量的应用,充分发挥 Flask 框架的潜力。
相关免费在线工具 curl 转代码 解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
Base64 字符串编码/解码 将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
Base64 文件转换器 将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
Markdown转HTML 将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online
HTML转Markdown 将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML转Markdown在线工具,online
JSON 压缩 通过删除不必要的空白来缩小和压缩JSON。 在线工具,JSON 压缩在线工具,online