参考资料
入门
安装
1
|
sudo pip3 install Flask requests
|
第一个 Flask 程序
创建项目结构
1
2
3
|
mkdir HelloWorld
mkdir HelloWorld/static
mkdir HelloWorld/templates
|
static 文件夹:存放静态文件,如图片、JS、CSS 文件
templates 文件夹:存放模板文件,如 HTML 文件
服务器端代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
# server.py
from flask import Flask
app = Flask(
'myapp-name', # app 名称
# static_folder='静态文件存放路径',
# template_folder='模板文件存放路径'
)
@app.route('/')
def hello_world():
return 'hello world!'
if __name__ == '__main__':
app.run(
host='0.0.0.0',
port=80,
debug=True,
)
|
关键说明:
- 当用户通过浏览器访问
/ 时,将看到 hello_world 函数返回的内容
- 这不仅仅是返回字符串,而是作为 HTTP 报文中的实体部分返回,状态码等信息由 Flask 自动处理
debug=True 开启调试模式,出现错误时会显示详细信息
host='0.0.0.0' 表示监听电脑的所有 IP 地址
port=80 是 HTTP 网站服务的默认端口,访问 http://www.example.com 时可省略端口号
- Flask 默认 IP 是
127.0.0.1,端口为 5000
请求处理
获取 URL 参数
URL 参数即出现在访问链接中的键值对。例如访问 http://www.example.com/?q=hello,URL 参数就是 {'q': 'hello'}。
1
2
3
4
5
6
7
8
9
10
11
12
|
from flask import Flask, request
app = Flask(__name__)
@app.route('/')
def hello_world():
print(request.path) # 路径部分,如 /
print(request.full_path) # 完整路径,如 /?q=hello&page=1
return request.args.__str__()
if __name__ == '__main__':
app.run(port=5000, debug=True)
|
访问 http://127.0.0.1:5000/?time&q=hello&page=1 时:
- 网页显示:
ImmutableMultiDict([('time', ''), ('q', 'hello'), ('page', '1')])
- 终端打印:
/ 和 /?time&q=hello&page=1
部分浏览器会自动将访问链接中的中文进行 URL 编码
获取指定参数:
1
2
|
request.args.get('info', 'hello') # 获取 info 参数,默认为 hello
request.args.getlist('info') # 获取参数的所有值,返回列表
|
获取 POST 方法传送的数据
POST 方法适用于以下场景:
- 注册账户,隐秘信息不希望显示在链接中
- 上传文章,文字内容可能导致链接参数过长
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from flask import Flask, request
app = Flask(__name__)
@app.route('/')
def hello_world():
return 'hello world!'
@app.route('/register', methods=['POST']) # 注意是 methods(复数)
def register():
print(request.headers) # 打印请求头
print(request.stream.read()) # 打印请求体数据
return 'welcome'
if __name__ == '__main__':
app.run(port=5000, debug=True)
|
请求头中的 Content-Type 为 application/x-www-form-urlencoded
POST 方法的参数解析:
request.form 解析请求体中的参数,用法类似 request.args:
1
2
3
|
request.form.get('name')
request.form.get('name', default='admin')
request.form.getlist('name') # 获取同名参数的多个值
|
客户端代码示例:
1
2
3
4
5
6
7
8
9
10
11
12
|
import requests
user_info = {
'name': 'Tom',
'passwd': 123
}
response = requests.post(
url='http://127.0.0.1:5000/register',
data=user_info
)
print(response.headers)
print(response.text)
|
JSON 格式数据
处理 JSON 格式的请求数据
通过 POST 方法传送的数据格式除了字典键值对外,还有 JSON、XML 等格式。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
from flask import Flask, request
app = Flask(__name__)
@app.route('/add', methods=['POST'])
def add():
print(request.headers)
# 自动将传入的 JSON 格式转换成 Python 字典
print(request.json)
result = request.json['name']
return result
if __name__ == '__main__':
app.run(port=5000, debug=True)
|
客户端代码:
1
2
3
4
5
6
7
8
9
10
11
|
import requests
json_data = {
'name': 'Tom',
'passwd': 123
}
response = requests.post(
url='http://127.0.0.1:5000/add',
json=json_data # 使用 json 参数
)
print(response.text)
|
请求头中的 Content-Type 为 application/json
服务器响应 JSON 数据
1
2
3
4
5
6
7
8
9
10
11
|
from flask import Flask, Response
import json
app = Flask(__name__)
@app.route('/add', methods=['POST'])
def add():
result = {'first name': request.json['name']}
resp = Response(json.dumps(result), mimetype='application/json')
resp.headers.add('Server', 'python flask') # 自定义响应头
return resp
|
也可以使用 Flask 的 jsonify 工具函数简化操作。
上传文件
文件上传一般使用 POST 方法,假设上传的文件存放在服务器端的 uploads 目录下。
1
2
|
mkdir HelloWorld/static/uploads
pip install werkzeug
|
服务器端代码:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
from flask import Flask, request
from werkzeug.utils import secure_filename
import os
app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = 'static/uploads/'
app.config['ALLOWED_EXTENSIONS'] = {'png', 'jpg', 'jpeg', 'gif'}
# app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 限制文件大小为 16MB
def allowed_file(filename):
return ('.' in filename) and \
(filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS'])
@app.route('/upload', methods=['POST'])
def upload():
upload_file = request.files['image']
if upload_file and allowed_file(upload_file.filename):
filename = secure_filename(upload_file.filename) # 检查文件名安全性
upload_file.save(os.path.join(
app.root_path, app.config['UPLOAD_FOLDER'], filename
))
return request.form.get('info', '') + 'success'
else:
return 'failed'
|
app.root_path 获取 server.py 所在目录的绝对路径
客户端代码:
1
2
3
4
5
6
7
8
|
import requests
response = requests.post(
url='http://127.0.0.1:5000/upload',
data={'info': 'baidu'},
files={'image': open('baidu.jpg', 'rb')}
)
print(response.text)
|
视图与路由
Restful URL
1
2
3
4
5
6
7
8
9
|
@app.route('/user/<username>')
def user(username):
print(username)
return 'hello ' + username
@app.route('/user/<username>/friends')
def user_friends(username):
print(username)
return 'hello ' + username + ', they are your friends!'
|
访问测试:
http://127.0.0.1:5000/user/tom → 显示 hello tom
http://127.0.0.1:5000/user/tom/friends → 显示 hello tom, they are your friends!
Restful URL 中的变量默认是 str 类型。转换类型可以使用:@app.route('/page/<int:num>')
还支持 float、path 等类型。
自定义转换类型可通过继承 werkzeug.routing.BaseConverter 实现。
URL 链接支持部分符号:
1
2
|
@app.route('/page/<int:num1>-<int:num2>')
# 访问 http://127.0.0.1:5000/page/11-22
|
url_for 反向路由
以软编码形式生成 URL(区别于 os.path.join 的硬编码方式):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
from flask import Flask, url_for
app = Flask(__name__)
@app.route('/')
def hello_world():
pass
@app.route('/user/<name>')
def user(name):
pass
@app.route('/page/<int:num>')
def page(num):
pass
@app.route('/test')
def test(username):
print(url_for('hello_world')) # /
print(url_for('user', name='tom')) # /user/tom
print(url_for('page', num=1, q='baidu google')) # /page/1?q=baidu+google
print(url_for('static', filename='uploads/01.jpg')) # /static/uploads/01.jpg
return ''
|
重定向
向客户端(浏览器)发送重定向的 HTTP 报文,浏览器会自动访问新 URL:
1
2
3
4
5
6
7
8
9
10
11
12
|
from flask import Flask, url_for, redirect
app = Flask(__name__)
@app.route('/test1')
def test1():
return redirect(url_for('test2'))
@app.route('/test2')
def test2():
print('This is test2.')
return 'This is test2.'
|
访问 http://127.0.0.1:5000/test1,浏览器 URL 会变成 http://127.0.0.1:5000/test2。
自定义错误响应
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from flask import Flask, abort, render_template_string
app = Flask(__name__)
@app.route('/user')
def user():
abort(401) # abort 后的代码不会执行
print('sorry')
@app.errorhandler(401)
def page_unauthorized(error):
tmp = render_template_string(
'<h1> Unauthorized </h1><h2>{{ error_info }}</h2>',
error_info=error
)
return tmp, 401
|
RESTful 架构
基本概念
REST(Representational State Transfer)是互联网软件的架构原则,中文译为"表现层状态转化"。
核心要点:
| 原则 |
说明 |
| 资源(Resources) |
网络上的一个实体,可以是一段文本、一张图片、一首歌曲、一种服务 |
| URI |
统一资源定位符,每个资源都有独一无二的地址 |
| 表现层(Representation) |
资源具体呈现的形式,如 txt、HTML、XML、JSON 格式 |
| 状态转化(State Transfer) |
通过 HTTP 协议的四个动词对资源进行操作 |
HTTP 动词:
| 动词 |
操作 |
| GET |
获取资源 |
| POST |
新建资源(也可用于更新资源) |
| PUT |
更新资源 |
| DELETE |
删除资源 |
RESTful 架构特征:
- 每一个 URI 代表一种资源
- 客户端和服务器之间传递资源的某种表现层
- 客户端通过四个 HTTP 动词,对服务器端资源进行操作
注意事项
1. URI 不应该有动词
错误的写法:
1
|
POST /accounts/1/transfer/500/to/2
|
正确的写法是把动词改成名词(服务资源):
1
2
3
4
|
POST /transaction HTTP/1.1
Host: 127.0.0.1
from=1&to=2&amount=500.00
|
2. 不要在 URI 中加入版本号
不同版本可以理解为同一种资源的不同表现形式,应该采用同一个 URI。版本号应在 HTTP 请求头的 Accept 字段中区分:
1
2
|
Accept: vnd.example-com.foo+json; version=1.0
Accept: vnd.example-com.foo+json; version=2.0
|
模板
Jinja2 模板引擎
Flask 默认使用 Jinja2 模板引擎,负责 MVC 中的 V(视图)部分。
模板继承
父模板 default.html:
1
2
3
4
5
6
7
8
|
<html>
<head>
<title>{% if page_title %} {{ page_title }} {% endif %}</title>
</head>
<body>
{% block body %}{% endblock %}
</body>
</html>
|
标签中的 if 判断用于条件渲染,block 用于被子模板继承填充。
子模板 user_info.html:
1
2
|
{% extends "default.html" %} {% block body %} {% for key in user_info %} {{ key
}}: {{ user_info[key] }} {% endfor %} {% endblock %}
|
user_info 变量应该是一个 Python 字典。
渲染模板
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from flask import Flask, render_template
app = Flask(__name__)
@app.route('/user')
def user():
user_info = {
'name': 'tom',
'age': '20',
'email': 'tom123@gmail.com'
}
return render_template(
'user_info.html',
page_title="tom's info", # 注意字符串中的引号处理
user_info=user_info
)
|
状态管理
用户会话(Session)
Session 用于记录用户的登录状态,一般基于 Cookie 实现。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
from flask import Flask, url_for, render_template_string, \
request, session, redirect
app = Flask(__name__)
app.secret_key = 'some_secret' # 用于给 session 加密
@app.route('/login')
def login():
page = '''
<form action="{{ url_for('do_login') }}" method="post">
<p>name: <input type="text" name="user_name" /></p>
<input type="submit" value="提交" />
</form>
'''
return render_template_string(page)
@app.route('/do_login', methods=['POST'])
def do_login():
name = request.form.get('user_name')
session['user_name'] = name
return 'success'
@app.route('/logout')
def logout():
session.pop('user_name', None) # 从 session 中移除用户
return redirect(url_for('login'))
@app.route('/show')
def show():
return session['user_name']
|
设置 Session 有效期:
1
2
3
4
|
from datetime import timedelta
session.permanent = True
app.permanent_session_lifetime = timedelta(minutes=5)
|
使用 Cookie
Cookie 是存储在客户端的记录访问者状态的数据。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
from flask import Flask, request, Response, make_response
import time
app = Flask(__name__)
@app.route('/add')
def login():
res = Response('add cookies')
# 设置 cookie 从现在开始的 6 分钟内有效
res.set_cookie(key='name', value='tom', expires=time.time() + 6*60)
return res
@app.route('/del')
def del_cookie():
res = Response('delete cookies')
res.set_cookie(key='name', value='', expires=0) # expires=0 删除 cookie
return res
@app.route('/show')
def show():
return request.cookies.__str__()
|
expires 参数可以是 datetime 对象或 Unix 时间戳
闪存系统
Flask 的闪存系统用于向用户提供反馈信息,这些信息是对上一次操作的反馈。反馈信息存储在服务器端,返回客户端后即被删除。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
from flask import Flask, flash, get_flashed_messages
import time
app = Flask(__name__)
app.secret_key = 'some_secret'
@app.route('/gen')
def gen():
info = 'access at ' + time.time().__str__()
flash(info)
return info
@app.route('/show')
def show():
return get_flashed_messages().__str__()
|
访问 /gen 后再访问 /show,可以看到之前 flash 的信息。一旦 show 过后,闪存信息就会被清空。
在模板中获取 flash 内容:
1
2
3
4
5
6
7
8
|
{% with messages = get_flashed_messages(with_categories=true) %} {% if messages
%}
<ul class="flashes">
{% for category, message in messages %}
<li class="{{ category }}">{{ message }}</li>
{% endfor %}
</ul>
{% endif %} {% endwith %}
|
蓝图
简介
蓝图的作用是将功能与主服务分开,把不同功能分到不同文件,相当于 Django 的子应用。
使用场景:
- 解决业务视图过多的问题
- 将业务单元划分出来单独维护
- 每个蓝图的视图、静态文件、模板文件都能独立分开
示例场景: 客户管理系统可以包含以下蓝图:
注意:蓝图不可以独立运行,必须注册到应用中才能使用。通常将蓝图对象创建在 __init__.py 文件中。
基本用法
项目结构:
1
2
3
4
5
|
displayCustomer/
- d_Cus.py
- template/
- static/
- app.py
|
蓝图定义 displayCustomer/d_Cus.py:
1
2
3
4
5
6
7
8
9
10
11
12
13
|
from flask import Blueprint
displayCus = Blueprint(
"displayCus", # 蓝图名字
__name__, # 蓝图所在位置
template_folder="template", # 蓝图模板文件目录
static_folder="static", # 蓝图静态文件目录
static_url_path="/s" # 蓝图静态文件访问路径
)
@displayCus.route("/displayCusList")
def view_Customer_List():
return "cuslist"
|
注册蓝图 app.py:
1
2
3
4
5
6
7
8
9
10
|
from flask import Flask
from displayCustomer import d_Cus
app = Flask(__name__)
app.register_blueprint(
d_Cus.displayCus,
url_prefix="/d" # 蓝图 URL 前缀
)
app.run("0.0.0.0", 5000)
|
蓝图可以暂时理解为没有 run 方法的 Flask 对象。每个蓝图都可以有独立的模板和静态文件目录。
注意事项:
- 蓝图模块中的视图函数路由不要出现重复
- Flask 会先去根目录下的 templates 文件夹中寻找模板文件,然后再去指定蓝图的模板文件目录中寻找
静态文件引用
在蓝图的模板文件中引用静态文件:
1
|
<link href="{{ url_for('blueprintName.static', filename='???.jpg') }}" />
|
使用 蓝图名.static 的方式,Flask 会去该蓝图指定的静态文件目录下寻找对应文件。
反转蓝图中的视图函数为 URL
1
|
url_for('blueprintName.index')
|
子域名
蓝图支持子域名配置:
1
2
|
displayCus = Blueprint("displayCus", __name__, subdomain="dpC")
app.config['SERVER_NAME'] = "flask_test.com:5000"
|
IP 地址和 localhost 不能有子域名。
需要添加域名与本机的映射:
| 系统 |
配置文件 |
| Windows |
C:\Windows\System32\drivers\etc\hosts |
| Linux |
/etc/hosts |
添加内容:
1
2
|
127.0.0.1 flask_test.com
127.0.0.1 dpC.flask_test.com
|
实战:售后数据可视化
项目结构
1
2
3
4
5
6
7
8
9
10
11
12
|
visualization/
- modules/
- configs/
- datafiles/
- data_process.py
- display.py
- static/
- css/
- js/
- templates/
- index.html
- app.py
|
前端依赖
1
2
|
<script type="text/javascript" src="static/js/myscript.js" defer></script>
<!-- defer 属性告诉浏览器在 HTML 文档解析完毕后再执行脚本 -->
|
第三方库:
- element-ui.css / element-ui.js
- plotly-latest.min.js
- jquery.min.js
- popper.min.js
- vue.js
- xm-select.js
后端代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
from flask import Flask, request, render_template, jsonify
from modules.data_process import DataProcess
from modules import display
import json
app = Flask(__name__, static_folder='static')
data_processor = None
@app.route('/', methods=['GET'])
@app.route('/index', methods=['GET'])
def index():
return render_template("index.html")
@app.route('/open_file', methods=['POST'])
def open_file():
global data_processor
excel_file = request.files['myFile']
try:
data_processor = DataProcess(excel_file.filename)
data_processor.main()
return jsonify({
'message': '文件处理完成',
'types': data_processor.select1,
'specifications': data_processor.select2,
'model': json.dumps(data_processor.model)
})
except Exception as e:
print(e)
# Flask 发生异常时默认返回 HTML 错误页面
# 将异常信息转换为字符串返回给前端
return str(e)
@app.route('/first_img', methods=['POST'])
def first_img():
x2 = request.form.get('x2').split(',')
stime = request.form.get('stime').split(',')
shower = display.show_images(data_processor.df, x2, startdate=stime[0], enddate=stime[1])
jsfig = shower.firstImg(colname='问题现象')
return jsonify({
'specification': x2[0],
'jsfig': jsfig
})
@app.route('/update_second', methods=['POST'])
def update_second():
specification = request.json['specification']
problem = request.json['problem']
stime = request.json['stime']
shower = display.show_images(data_processor.df, specification, problem, startdate=stime[0], enddate=stime[1])
jsfig = shower.secondImg(colname='问题原因')
return jsonify({'jsfig': jsfig})
@app.route('/update_third', methods=['POST'])
def update_third():
specification = request.json['specification']
problem = request.json['problem']
reason = request.json['reason']
stime = request.json['stime']
shower = display.show_images(data_processor.df, specification, problem, reason, startdate=stime[0], enddate=stime[1])
jsfig = shower.thirdImg(colname='解决方案')
return jsonify({'jsfig': jsfig})
if __name__ == '__main__':
app.run(debug=True)
|
运行方式:
1
2
3
4
|
# 设置临时 PYTHONPATH
$env:PYTHONPATH="D:\jupyterproject\...\visualization"
cd visualization
python app.py
|
拓展与后续学习建议
延伸阅读方向:
- Flask 扩展生态:深入学习 Flask-SQLAlchemy、Flask-Migrate、Flask-Login 等常用扩展
- RESTful API 开发:结合 Flask-RESTful 或 Flask-RESTX 构建规范化的 API 服务
- 异步任务处理:学习 Celery + Redis 实现异步任务队列
高阶应用场景:
- 项目结构工程化:探索大型 Flask 项目的目录组织方式,如 Blueprints 分层设计
- API 文档生成:集成 Flask-APISpec 或 Swagger 实现自动 API 文档
- 部署与容器化:学习使用 Gunicorn/uWSGI + Nginx 部署 Flask 应用,结合 Docker 容器化
- 数据库进阶:掌握 SQLAlchemy 的更多特性,包括连接池管理、事务控制、查询优化等