flask学习
更新时间: 2025-12-13 12:56:22
# 初识Flask
# 安装Flask
conda install flask # 通过 Conda 仓库安装（版本可能滞后）
# 或使用 pip（推荐，获取最新版）
pip install flask
1
2
3
2
3
当我们把项目创建完了，并且把flask创建成功，我们接下来就回到今天的主题
我们写一个基于flask的网站，来接收用户请求，并且获取数据并且给它返回结果
# 初识flask
首先引入flask，这是个固定写法
from flask import Flask
app = Flask(__name__)
1
2
3
2
3
然后我们定义一个函数，在上面写上@app.route("/xxx"),我们在浏览器上访问/xxx的时候，就会执行这个index函数，然后把返回值给用户
# /xxx -> 执行index函数
@app.route('/xxx')
def index():
return "成功"
# /yyy -> 执行home函数
@app.route('/yyy')
def home():
return "失败"
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
运行网站
if __name__ == '__main__':
app.run()
1
2
2
运行一下代码
代码直接就启动起来了
这跟以前写python脚本不一样，而这相当于写了个网站，我们在等待浏览器的访问
基于flask，这几行代码就相当于写了个网站出来
注意
- 一般情况下url的名字都会和函数名一致，虽然他们本身没有关系，但这样我们找函数名的时候会比较好找
下面这样相当于指定ip和端口
app.run(host="127.0.0.1",port=5000)
1
写完以后如果想让代码生效，写完后必须先让之前运行的程序终止，再运行一下
# 获取请求数据
# get请求
有时候用户会在url上传入参数，比如 http://127.0.0.1:5000/index?age=19&pwd=123
我们需要多引入一个request
from flask import Flask,request
app = Flask(__name__)
# /xxx -> 执行index函数
# http://127.0.0.1:5000/index?age=19&pwd=123
@app.route('/xxx')
def index():
age = request.args.get("age")
pwd = request.args.get("pwd")
return "成功"+age+"|"+pwd
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
然后我们访问一下试试，记得要重启哦
# post请求
默认情况下是不支持post请求的，如果需要支持POST请求需要配置一下
@app.route('/xxx', methods=['POST','GET'])
1
post请求会在请求体里携带数据,可以这样获取到
request.form.get("age")
1
如果请求体是json格式的比如
{
"xx":123,
"yy":999
}
1
2
3
4
2
3
4
就用
request.json
1
# 返回json数据
一般情况下，我们会返回一个json格式的数据
from flask import Flask,request,jsonify
app = Flask(__name__)
# /xxx -> 执行index函数
# http://127.0.0.1:5000/index?age=19&pwd=123 -> 执行 get
# http://127.0.0.1:5000/index -> 执行 post
@app.route('/xxx', methods=['POST','GET'])
def index():
age = request.args.get("age")
pwd = request.args.get("pwd")
return jsonify({"age":age,"pwd":pwd})
1
2
3
4
5
6
7
8
9
10
11
12
13
2
3
4
5
6
7
8
9
10
11
12
13
# 直接访问API并返回结果
# hashlib是Python标准库中提供加密哈希算法的模块，用于实现数据的不可逆加密和完整性校验。它整合了多种安全哈希算法，提供统一的API接口，是处理密码存储、数据校验和数字签名的核心工具。
import hashlib
from flask import Flask,request,jsonify
app = Flask(__name__)
@app.route('/bili', methods=['POST'])
def bili():
"""
请求的数据格式要求 ：{"ordered_string":"....."}
:return:
"""
ordered_string = request.json.get('ordered_string')
if not ordered_string:
return jsonify({'error':'参数错误', 'status': False})
# 调用核心算法，生成sign签名
encrypt_string = ordered_string + "560c52ccd288fed045859ed18bffd973"
obj = hashlib.md5(encrypt_string.encode("utf-8"))
sign = obj.hexdigest()
return jsonify({"status":True, "data":sign})
if __name__ == '__main__':
app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
我们返回一个加密过的sign，然后返回json格式
现在用postman测试一下看看可不可以
有时候我们是买的别人的服务，要如何调用别人的服务呢
import requests
res = requests.post(
url="http://127.0.0.1:5000/bili",
json = {
"ordered_string": "123"
}
)
print(res.json())
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
::: tips 无校验 直接提供API，接收用户传来的请求，根据请求执行自己的签名，直接返回用户结果
- 安装Flask
- 编写路由和视图
- 接收用户请求： GET/POST传入的数据
- 返回json数据
:::
# 基于文件进行授权
如果这样的话，那代表任何人都可以向我们的地址发送POST请求，都可以获取到相应的签名，这样肯定是不合理的
所以这个时候，我们就得做上一个授权，只有通过授权的人，才可以访问
首先我们先生成一个随机的uuid
import uuid
print(uuid.uuid4())
1
2
3
2
3
然后生成几个凭证仅仅允许这一个用户访问,我们把凭证放在db.txt中
8eb2d5c6-88d9-4006-bb6e-8315194c3a98,李鹏
c9be75b8-3362-4772-ad54-8d428556eb24,李狗
1
2
2
然后执行接口之前先进行凭证的校验
# hashlib是Python标准库中提供加密哈希算法的模块，用于实现数据的不可逆加密和完整性校验。它整合了多种安全哈希算法，提供统一的API接口，是处理密码存储、数据校验和数字签名的核心工具。
import hashlib
from flask import Flask,request,jsonify
app = Flask(__name__)
def get_user_dict():
info_dict = {
}
with open("db.txt", mode="r",encoding="utf-8") as f:
for line in f:
line = line.strip()
token, name = line.split(",")
info_dict[token] = name
return info_dict
@app.route('/bili', methods=['POST'])
def bili():
"""
请求的url中需要携带token /bili?token=xxx
请求的数据格式要求 ：{"ordered_string":"....."}
:return:
"""
token = request.args.get('token')
if not token:
return jsonify({'status':False,'error':'认证失败'})
user_dict = get_user_dict()
if token not in user_dict:
return jsonify({'status':False,'error':'认证失败'})
ordered_string = request.json.get('ordered_string')
if not ordered_string:
return jsonify({'error':'参数错误', 'status': False})
# 调用核心算法，生成sign签名
encrypt_string = ordered_string + "560c52ccd288fed045859ed18bffd973"
obj = hashlib.md5(encrypt_string.encode("utf-8"))
sign = obj.hexdigest()
return jsonify({"status":True, "data":sign})
if __name__ == '__main__':
app.run()
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
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
之所以要写到文件里面是因为我们网站运行就不需要再停止了，如果用户信息写在python代码里，那么来一个新的用户就要重新重启一下网站，这样是不行的
# 数据库版本授权
刚刚我们把授权相关的信息存储文件里了，但是写文件里操作不太方便，所以我们把它放到mysql数据库里
首先在数据库里建一个相关的表
再创建一张表
下面在程序里连接上数据库
# hashlib是Python标准库中提供加密哈希算法的模块，用于实现数据的不可逆加密和完整性校验。它整合了多种安全哈希算法，提供统一的API接口，是处理密码存储、数据校验和数字签名的核心工具。
import hashlib
from flask import Flask,request,jsonify
import pymysql
app = Flask(__name__)
@app.route('/bili', methods=['POST'])
def bili():
"""
请求的url中需要携带token /bili?token=xxx
请求的数据格式要求 ：{"ordered_string":"....."}
:return:
"""
# 1. token 是否为空
token = request.args.get('token')
if not token:
return jsonify({'status':False,'error':'认证失败'})
# 2.token是否合法，连接MySQL执行命令
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='tree930827', db='test', charset='utf8')
cursor = conn.cursor()
cursor.execute("select * from user where token=%s",[token])
result = cursor.fetchone()
cursor.close()
conn.close()
if not result:
return jsonify({'status':False,'error':'认证失败'})
ordered_string = request.json.get('ordered_string')
if not ordered_string:
return jsonify({'error':'参数错误', 'status': False})
# 调用核心算法，生成sign签名
encrypt_string = ordered_string + "560c52ccd288fed045859ed18bffd973"
obj = hashlib.md5(encrypt_string.encode("utf-8"))
sign = obj.hexdigest()
return jsonify({"status":True, "data":sign})
if __name__ == '__main__':
app.run()
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
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
然后给数据库里添加一些数据
然后请求接口
# 集成MySQL数据库连接池
按照我们刚才的方式，我们每次都要跟数据库连接数据库，其实这种方式性能会比较低，而真正在公司做开发的时候，也不会这样每次都连接数据库，性能损耗比较严重。
但如果用同一个连接，中间有多个请求还会有并发问题。
所以我们可以用连接池这个机制，在python中想使用连接池，就需要安装dbutils
# hashlib是Python标准库中提供加密哈希算法的模块，用于实现数据的不可逆加密和完整性校验。它整合了多种安全哈希算法，提供统一的API接口，是处理密码存储、数据校验和数字签名的核心工具。
import hashlib
from flask import Flask,request,jsonify
import pymysql
from dbutils.pooled_db import PooledDB
app = Flask(__name__)
POOL = PooledDB(
creator=pymysql, # 使用链接数据库的模块
maxconnections=10, # 连接池允许的最大连接数，0和None表视不限制连接数
mincached=2, # 初始化时，链接池中至少创建的空闲的连接，0表示不创建
maxcached=5, # 链接池中最多闲置的链接，0和None表示不限制
blocking=True, # 链接池中如果没有可用连接后，是否阻塞等待。True,等待；False,不等待然后报错
setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...","set time zone..."]
ping=0,
host='127.0.0.1', # 剩下的就是数据库连接用的东西了，这几个东西会被传给pymysql
port=3306,
user='root',
passwd='tree930827',
db='test',
charset='utf8'
)
@app.route('/bili', methods=['POST'])
def bili():
"""
请求的url中需要携带token /bili?token=xxx
请求的数据格式要求 ：{"ordered_string":"....."}
:return:
"""
# 1. token 是否为空
token = request.args.get('token')
if not token:
return jsonify({'status':False,'error':'认证失败'})
# 2.token是否合法，连接MySQL执行命令
conn = POOL.connection()
cursor = conn.cursor()
cursor.execute("select * from user where token=%s",[token])
result = cursor.fetchone()
cursor.close()
conn.close() # 这里不是关闭连接了，将此连接交还给链接池
if not result:
return jsonify({'status':False,'error':'认证失败'})
ordered_string = request.json.get('ordered_string')
if not ordered_string:
return jsonify({'error':'参数错误', 'status': False})
# 调用核心算法，生成sign签名
encrypt_string = ordered_string + "560c52ccd288fed045859ed18bffd973"
obj = hashlib.md5(encrypt_string.encode("utf-8"))
sign = obj.hexdigest()
return jsonify({"status":True, "data":sign})
if __name__ == '__main__':
app.run()
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
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
# 接收任务并加入队列
前面我们都是做的不耗时的，但如果我们给别人处理需要很长时间才能执行出来，那么可以基于这种来做
- 调用者：携带参数发送请求
- API：接收请求并为请求生成一个任务ID,接下来：返回给调用者+放到任务队列
- worker: 等待redis队列（List）,一旦接到任务，就执行并将结果返回到结果队列（Hash）
- 调用者：等待n秒后，携带任务ID再次发起请求，获取结果。
- APU：接受任务ID，根据ID去结果队列（hash）中获取
下载安装redis https://github.com/tporadowski/redis/releases (opens new window)
下载zip版的直接解压就能用
# 安装并测试redis
首先安装redis
pip install redis
1
运行起来redis
import redis
REDIS_CONN_PARAMS = {
'host': '127.0.0.1',
'port': 6379,
'encoding': 'utf-8'
}
conn = redis.Redis(**REDIS_CONN_PARAMS)
conn.lpush("test_spider_task_list", "123")
conn.lpush("test_spider_task_list", "456")
data = conn.rpop("test_spider_task_list")
print(data)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
2
3
4
5
6
7
8
9
10
11
12
13
14
试一下，我们取出的data是123，redis现在正常能用
因此我们的接口去掉token先把消息队列加上
# hashlib是Python标准库中提供加密哈希算法的模块，用于实现数据的不可逆加密和完整性校验。它整合了多种安全哈希算法，提供统一的API接口，是处理密码存储、数据校验和数字签名的核心工具。
import hashlib
import uuid
import json
from flask import Flask,request,jsonify
import redis
app = Flask(__name__)
@app.route('/task', methods=['POST'])
def task():
"""
请求的数据格式要求 ：{"ordered_string":"....."}
:return:
"""
token = request.args.get('token')
if not token:
return jsonify({'status':False,'error':'认证失败'})
ordered_string = request.json.get('ordered_string')
if not ordered_string:
return jsonify({'error': '参数错误', 'status': False})
# 生成任务ID
tid = str(uuid.uuid4())
# 1.放入到队列中
task_dict = {'tid':tid, 'data': ordered_string}
REDIS_CONN_PARAMS = {
'host': '127.0.0.1',
'port': 6379,
'encoding': 'utf-8'
}
conn = redis.Redis(**REDIS_CONN_PARAMS)
conn.lpush("spider_task_list", json.dumps(task_dict))
# 2.给用户返回
return jsonify({"status":True, "data":tid, 'message': '正在处理中，预计1分钟完成'})
if __name__ == '__main__':
app.run()
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
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
我们现在直接访问task接口
