Flask——接口路由技术

一、Flask 简介

Flask 是一个轻量级的 web 开发框架。 它依赖 jinja2 和 Werkzeug WSGI 服务的一个微型框架。

1、环境安装：

pip install flask

2、一个最小的应用

# 1. 导入 Flask 模块
from flask import Flask
# 2. 创建Flask应用程序实例
app = Flask(__name__)

# 3. 定义路由及视图函数
@app.route("/")
def hello_world():
    return "
Hello, World!
"
# 4. 启动程序
if __name__ == '__main__':
    app.run()
1
2
3
4
5
6
7
8
9
10
11
12

3、两种运行方式

linux/mac: 命令行运行

$ export FLASK_APP=hello
$ flask run

windows: 命令运行

set FLASK_APP=app.py
flask run

代码调用

if name == ‘main’:
app.run()

二、定义路由

1、普通路由

通过装饰器 @app.route 添加路由

from flask import Flask

#2、创建flask应用程序的实例；__name__==__main__
app=Flask(__name__)

#添加路由
@app.route("/")
def hello_world():
    return "
hello world
"

@app.route("/demo")
def demo():
    return "
hello world 666
"

#启动入口
if __name__=='__main__':
    #flask服务启动起来
    #轮询等待的方式，等待浏览器发来请求
    #会一直接收请求，直到程序停止
    app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2、动态路由

通过 app.route('/user/') 添加动态路由

from flask import Flask

#2、创建flask应用程序的实例；__name__==__main__
app=Flask(__name__)

#添加路由
@app.route("/")
def hello_world():
    return "
hello world
"

@app.route("/demo")
def demo():
    return "
hello world 666
"

@app.route("/userinfo/")
def demo1(username):
    return f"hello world 这是{username}同学"

#启动入口
if __name__=='__main__':
    #flask服务启动起来
    #轮询等待的方式，等待浏览器发来请求
    #会一直接收请求，直到程序停止
    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

3、限定类型

路径中添加 <类型：变量名> 来限定变量的类型
@app.route('/post/')

string：接收任何不包含斜杠的文本
int：接收正整数
float：接收正浮点数
path：类似string，但可以包含斜杠
uuid：接收UUID字符串

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/5/26 07:03
# @Author  : 杜兰特
# @File    : flask_demo.py

#1、导入flask模块
from log_utils import logger

from flask import Flask

#2、创建flask应用程序的实例；__name__ ==__main__
app=Flask(__name__)

#限定类型
@app.route("/user/")
def demo2(id):
    return f"hello world 这是{id}"

@app.route("/user/")
def demo3(username):
    return f"hello world 这是{username}同学"

#启动入口
if __name__=='__main__':
    #flask服务启动起来
    #轮询等待的方式，等待浏览器发来请求
    #会一直接收请求，直到程序停止
    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

4、地址尾部的“/”

路由的尾部带有“/”（浏览器的地址栏中输入和不输入“/”的效果一样）

路由的尾部没有“/”（输入的 URL 的结尾不能加“/”，会报错）

#1、导入flask模块
from flask import Flask

#2、创建flask应用程序的实例；__name__ ==__main__
app=Flask(__name__)

#添加路由
@app.route("/demo")
def demo():
    return "
hello world 666
"

#启动入口
if __name__=='__main__':
    #flask服务启动起来
    #轮询等待的方式，等待浏览器发来请求
    #会一直接收请求，直到程序停止
    app.run()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

三、请求与响应-请求方法

请求 说明

GET 获取服务器资源
POST 新增服务器资源
PUT 更新服务器资源（客户端提供改变后的完整资源）
DELETE 删除服务器资源

接口设计

查询：get
新增：post
修改：put
删除：delete

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/5/27 00:01
# @Author  : 杜兰特
# @File    : http_request_demo.py
from flask import Flask

app=Flask(__name__)

#methods是列表类型，可以添加多种请求方式。get、post、put、delete
@app.route("/cases",methods=['get'])
def get_case():
    return {"code":0,"msg":"get success"}

@app.route("/cases",methods=['post'])
def post_case():
    return {"code":0,"msg":"post success"}

@app.route("/cases",methods=['put'])
def put_case():
    return {"code":0,"msg":"put success"}

@app.route("/cases",methods=['delete'])
def delete_case():
    return {"code":0,"msg":"delete success"}

if __name__ == '__main__':
    app.run(debug=True)
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

四、请求与响应-处理请求数据

1、request的常用属性/方法

args：记录请求中的查询参数
json：记录请求中的json数据
files：记录请求上传的文件
form：记录请求中的表单数据
method：记录请求使用的http方法
url：记录请求的URL地址
host：记录请求的域名
headers：记录请求的头信息

2、get 请求参数

场景：普通的 url 链接，接收一个 get 请求
解决办法：request.args

#需要导入request，而不是requests！！！
from flask import Flask,request
from log_utils import logger

app=Flask(__name__)

@app.route('/login',methods=['get'])
def demo1():
    logger.info(f"请求参数为：{request.args}")
    a=request.args.get("a")
    b=request.args.get("b")
    logger.info(f"a的值为{a},b的值为{b}")
    return {"code":0,"msg":"get success"}

if __name__ == '__main__':
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

127.0.0.1 - - [27/May/2023 09:03:29] "GET /login HTTP/1.1" 200 -
[2023-05-27 09:04:04] [INFO] [flask_param_demo.py]/[line: 17]/[demo1] 请求参数为：ImmutableMultiDict([('a', '88'), ('b', '66')]) 
[2023-05-27 09:04:04] [INFO] [flask_param_demo.py]/[line: 20]/[demo1] a的值为88,b的值为66 
1
2
3

3、json 请求

场景：POST 相关的请求，带有 json 数据格式
解决办法：request.json

#需要导入request，而不是requests！！！
from flask import Flask,request
from log_utils import logger

app=Flask(__name__)

@app.route('/register',methods=['post'])
def demo2():
    logger.info(f"请求参数为：{request.json}")
    data=request.json.get("time")
    logger.info(f"时间为{data}")
    return {"code":0,"msg":"post success"}

if __name__ == '__main__':
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

[2023-05-27 09:05:55] [INFO] [flask_param_demo.py]/[line: 26]/[demo2] 请求参数为：{'time': '2023-05-27', 'status': 'success'} 
[2023-05-27 09:05:55] [INFO] [flask_param_demo.py]/[line: 28]/[demo2] 时间为2023-05-27 
1
2

4、表单请求

场景：比如：测试人网站的登录接口，需要用户名和密码，前端会提交一个 form 表单给后台
解决办法：request.form

#需要导入request，而不是requests！！！
from flask import Flask,request
from log_utils import logger

app=Flask(__name__)

@app.route('/register',methods=['put'])
def demo3():
    logger.info(f"请求参数为：{request.form}")
    name=request.form.get("name")
    password=request.form.get("password")
    email=request.form.get("email")
    logger.info(f"请求参数name为：{name},password为：{password}，email为；{email}")
    return {"code": 0, "msg": "put success"}

if __name__ == '__main__':
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

[2023-05-27 09:07:42] [INFO] [flask_param_demo.py]/[line: 34]/[demo3] 请求参数为：ImmutableMultiDict([('name', 'kobe'), ('password', '666666'), ('email', '77777')]) 
[2023-05-27 09:07:42] [INFO] [flask_param_demo.py]/[line: 38]/[demo3] 请求参数name为：kobe,password为：666666，email为；77777 
1
2

5、文件请求

场景：
页面上有个更新头像的功能， 或者上传一个 excel 文件的功能， 允许我们提交一个图片，或者文件到后端服务器，那么
解决方法：
request.files.get(‘file’) 获取文件对象
filename 获取文件对象的文件名
save()方法 保存文件到指定路径下

#需要导入request，而不是requests！！！
from flask import Flask,request
from log_utils import logger


app=Flask(__name__)

@app.route('/file',methods=['post'])
def demo4():
    logger.info(f"请求方法为：{request.method}")
    logger.info(f"请求url为：{request.url}")
    logger.info(f"请求host为：{request.host}")
    logger.info(f"请求headers为：{request.headers}")
    fileobj=request.files.get("name")
    logger.info(f"fileobj的值为：{fileobj}")
    logger.info(f"文件名为：{fileobj.filename}")
    #保存文件
    fileobj.save(f"./{fileobj.filename}")
    return {"code": 0, "msg": "file success"}

if __name__ == '__main__':
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

[2023-05-27 09:10:10] [INFO] [flask_param_demo.py]/[line: 43]/[demo4] 请求方法为：POST 
[2023-05-27 09:10:10] [INFO] [flask_param_demo.py]/[line: 44]/[demo4] 请求url为：http://127.0.0.1:5000/file 
[2023-05-27 09:10:10] [INFO] [flask_param_demo.py]/[line: 45]/[demo4] 请求host为：127.0.0.1:5000 
[2023-05-27 09:10:10] [INFO] [flask_param_demo.py]/[line: 46]/[demo4] 请求headers为：User-Agent: PostmanRuntime/7.32.2
Accept: */*
Postman-Token: 593d536e-4bbd-45a2-89be-d8670f66aa42
Host: 127.0.0.1:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
Content-Type: multipart/form-data; boundary=--------------------------624583842625202752710421
Content-Length: 10016

 
[2023-05-27 09:10:10] [INFO] [flask_param_demo.py]/[line: 48]/[demo4] fileobj的值为：<FileStorage: '员工用车导入模板 (2).xlsx' ('application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')> 
[2023-05-27 09:10:10] [INFO] [flask_param_demo.py]/[line: 49]/[demo4] 文件名为：员工用车导入模板 (2).xlsx 
127.0.0.1 - - [27/May/2023 09:10:10] "POST /file HTTP/1.1" 200 -

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

五、请求与响应-处理响应信息

1、文本型

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/5/27 09:27
# @Author  : 杜兰特
# @File    : http_response_demo.py

from flask import Flask
from flask import jsonify,render_template,make_response

app=Flask(__name__)

#返回文本
@app.route("/text")
def demo():
    return "文本信息"

if __name__ == '__main__':
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

2、元组

返回元组
(response, status)
(response, headers)
(response, status, headers)
响应状态码默认为 200

from flask import Flask
from flask import jsonify,render_template,make_response

app=Flask(__name__)


#返回元组
@app.route("/tuple")
def tuple_res():
    return 'hello',200,{"status": 0, "type": "success"}


if __name__ == '__main__':
    #host
    #port
    #debug=True:热加载，方便调试
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3、Json

直接返回 dict 会转换为 json
使用jsonify()方法，通过参数传入键值对

from flask import Flask
from flask import jsonify,render_template,make_response

app=Flask(__name__)

#返回字典
@app.route("/dict")
def get_dict_res():
    return {"status":0}

@app.route("/json")
def get_json_res():
    #return jsonify({"status":0})
    #或者
    return jsonify(status=1,name="kobe")

if __name__ == '__main__':
    #host
    #port
    #debug=True:热加载，方便调试
    app.run(debug=True)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

4、html

使用模板渲染技术
html 文件必须在同级的 templates 目录下

from flask import Flask
from flask import jsonify,render_template,make_response

app=Flask(__name__)


@app.route("/html")
def get_html():
    return render_template("demo.html")


if __name__ == '__main__':
    #host
    #port
    #debug=True:热加载，方便调试
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

5、额外数据

设置额外数据-make_response()
添加更多的响应信息

设置 cookie
设置响应头信息等

from flask import Flask
from flask import jsonify,render_template,make_response

app=Flask(__name__)

@app.route("/")
def index():
    resp=make_response(render_template("demo.html"))
    #设置cookie
    resp.set_cookie("user","jimi")
    #设置响应头信息
    resp.headers["type"]="kobe"
    return resp

if __name__ == '__main__':
    #host
    #port
    #debug=True:热加载，方便调试
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

六、测试平台环境配置

监听的主机

设置 host 参数
127.0.0.1 只能本机访问
0.0.0.0 服务发布到局域网

app.run(host="0.0.0.0")
1

监听的端口
设置 port 参数（默认端口号 5000）

app.run(host="0.0.0.0", port=5000)
1

Debug 模式
设置 debug=True（默认是 production）

实现热加载
开发调试方便

app.run(host="0.0.0.0", port=5000,debug=True)
1

七、接口配置

1、flask-restx 介绍

官方文档：https://github.com/python-restx/flask-restx
flask-restx 是一个支持 RESTFUL 的 flask 插件，用于规范化接口的编写，并且支持 swagger 文档。

2、插件安装

pip install flask-restx
1

3、demo 示例

from flask import Flask
#from flask_restful import Resource,Api
from flask_restx import Resource,Api

app=Flask(__name__)

#1、创建api实例对象
api=Api(app)

#2、使用api添加路由
@api.route('/hello')
class HelloWorld(Resource):     #3、类要继承Resource
    #4、定义restful风格的get方法
    def get(self):
        return {'hello':'world'}

    # restful风格的post方法
    def post(self):
        return {'post':'success'}

if __name__ == '__main__':
    app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

八、接口使用

1、设计框架的原则

复用性
高内聚，低藕合

a、高耦合实例

问题：判断条件过多，业务逻辑非常复杂

@app.route("/testcase",methods=["GET","PUT","POST"])
def select_case():
    ### request:Request
    if request.method == "GET":
        pass
    elif request.method == "PUT":
        pass
1
2
3
4
5
6
7

b、低内聚示例

问题：同一个路径，对应多个请求方法，代码没有复用

# get 请求
@app.route("/testcase",methods=["get"])
def select_case():
    # request:Request
    # m = request.method
    print(dir(request))
    return {"code": 0, "msg":"get success"}

# post 请求
@app.route("/testcase",methods=["post"])
def post_case():
    return {"code": 0, "msg":"post success"}

# delete 请求
@app.route("/testcase",methods=["delete"])
def delete_case():
    return {"code": 0, "msg":"delete success"}


# put 请求
@app.route("/testcase",methods=["put"])
def put_case():
    return {"code": 0, "msg":"put success"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

2、编写 RESTFUL 风格的接口

导入from flask_restx import Resource,Api
app=Flask(name) 创建app实例对象
api=Api(app) 创建api实例对象
使用api添加路由
类要继承Resource
定义restful风格的方法

from flask import Flask
#from flask_restful import Resource,Api
from flask_restx import Resource,Api

app=Flask(__name__)

#1、创建api实例对象
api=Api(app)

#2、使用api添加路由
#@api.route('/user','/user1','/user2')  添加多个路由
@api.route('/user')
class User(Resource):     #3、类要继承Resource
    #4、定义restful风格的get方法
    def get(self,id):
        return {"id":id,'code':0,"msg":"get success"}

    # restful风格的post方法
    def post(self):
        return {'code':0,"msg":"post success"}

    # restful风格的put方法
    def put(self):
        return {'code':0,"msg":"put success"}

    # restful风格的delete方法
    def delete(self):
        return {'code':0,"msg":"delete success"}


api.add_resource(User,'/user/','/user_1')

if __name__ == '__main__':
    app.run(debug=True)
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

3、添加路由的方式

方式一：

添加多个路由
@api.route('/user','/user1','/user2')

@api.route('/user','/user1','/user2')  添加多个路由
@api.route('/user')
class User(Resource):
1
2
3

方式二

api.add_resource(User,'/user/','/user_1')

九、集成swagger

1、namespace 的使用

定义 Namespace 实例
为类添加装饰器 @namespace.route(“”) 控制子路由
为命名空间指定访问资源路径 api.add_namespace(case_ns, ‘/case’)

from flask import Flask
from flask_restx import Api,Resource,Namespace

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

#定义了2个命名空间
hello_ns=Namespace("demo",description="demo学习")
case_ns=Namespace("case",description="用例管理")

#将api.route("case")  改为：case_ns.route("/case")
#@case_ns.route("") 定义子路由，如果没有的话，传空字符串即可
@case_ns.route("")
class TestCase(Resource):

    def get(self):
        pass

    def post(self):
        pass

    def put(self):
        pass

    def delete(self):
        pass


@hello_ns.route("/demo")	#子路由
class Demo(Resource):

    def get(self):
        pass

    def post(self):
        pass

    def put(self):
        pass

    def delete(self):
        pass

api.add_namespace(hello_ns,"/hello")
api.add_namespace(case_ns,"/case")

if __name__ == '__main__':
    app.run(debug=True)
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

2、swagger 接口文档配置

定义参数 parser = api.parser()
传递及校验参数 @api.expect(parser) 或者 @namespace.expect(parser)

a、api.parser()用法

格式：api.parser().add_argument(参数名, 关键字参数)
第一个参数是参数名
后面是关键字传参，常用的关键字有：
type ：类型
required 约束控制
choices 枚举参数
location 对应 request 对象中的属性

b、get 请求示例

from flask import Flask, request
from flask_restx import Resource, Api, Namespace,fields
from log_utils import logger
from werkzeug.datastructures import FileStorage

app = Flask(__name__)
api = Api(app)
hello_ns = Namespace("demo", description="demo学习")

@hello_ns.route("")
class Demo(Resource):
    #定义parser解析器对象
    get_parser = api.parser()
    #通过parser对象添加测试参数
    get_parser.add_argument('id',
                            type=int,
                            location="args",
                            required=True)

    get_parser.add_argument('case_title',
                            type=str,
                            location="args",
                            required=True)

    @hello_ns.expect(get_parser)
    def get(self):
        logger.info(f"request.args ===>{request.args}")
        return {"code": 0, "msg": "get success"}


api.add_namespace(hello_ns,"/hello")

if __name__ == '__main__':
    app.run(debug=True)
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

接口文档如下

c、post 请求示例

1、处理 json 格式

from flask import Flask, request
from flask_restx import Resource, Api, Namespace,fields
from log_utils import logger
from werkzeug.datastructures import FileStorage

app = Flask(__name__)
api = Api(app)
hello_ns = Namespace("demo", description="demo学习")

@hello_ns.route("")
class Demo(Resource):
    
    post_parser=api.parser()
    post_parser.add_argument("id",type=int,location="json",required=True)
    post_parser.add_argument("casetitle",type=str,location="json",required=True)


    @hello_ns.expect(post_parser)
    def post(self):
        logger.info(f"request.json ===>{request.json}")
        return {"code": 0, "msg": "post success"}


api.add_namespace(hello_ns,"/hello")

if __name__ == '__main__':
    app.run(debug=True)
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

2、处理 files格式

from flask import Flask, request
from flask_restx import Resource, Api, Namespace,fields
from log_utils import logger
from werkzeug.datastructures import FileStorage

app = Flask(__name__)
api = Api(app)
hello_ns = Namespace("demo", description="demo学习")

@hello_ns.route("")
class Demo(Resource):
    
    post_parser=api.parser()
    post_parser.add_argument("file",type=FileStorage,location="files",required=True)

    @hello_ns.expect(post_parser)
    def post(self):
        logger.info(f"request.files ===>{request.files}")
        return {"code": 0, "msg": "post success"}


api.add_namespace(hello_ns,"/hello")

if __name__ == '__main__':
    app.run(debug=True)
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

请求接口

3、处理 form格式

from flask import Flask, request
from flask_restx import Resource, Api, Namespace,fields
from log_utils import logger
from werkzeug.datastructures import FileStorage

app = Flask(__name__)
api = Api(app)
hello_ns = Namespace("demo", description="demo学习")

@hello_ns.route("")
class Demo(Resource):
    
    post_parser=api.parser()
    post_parser.add_argument("param1",help="username",type=int,location="form",required=True)
    post_parser.add_argument("param2",help="password",type=int,location="form",required=True)

    @hello_ns.expect(post_parser)
    def post(self):
        logger.info(f"request.form ===>{request.form}")
        return {"code": 0, "msg": "post success"}


api.add_namespace(hello_ns,"/hello")

if __name__ == '__main__':
    app.run(debug=True)
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

[2023-05-28 22:04:04] [INFO] [swagger_demo2.py]/[line: 53]/[post] request.form ===>ImmutableMultiDict([('param1', '24'), ('param2', '123567')]) 
127.0.0.1 - - [28/May/2023 22:04:04] "POST /hello HTTP/1.1" 200 -
1
2

4、处理choice格式

from flask import Flask, request
from flask_restx import Resource, Api, Namespace,fields
from log_utils import logger
from werkzeug.datastructures import FileStorage

app = Flask(__name__)
api = Api(app)
hello_ns = Namespace("demo", description="demo学习")

@hello_ns.route("")
class Demo(Resource):
    
    post_parser=api.parser()
    post_parser.add_argument("choice", location="args", required=True,choices=(1,2,3,4))

    @hello_ns.expect(post_parser)
    def post(self):
        logger.info(f"request.args ===>{request.args}")
        return {"code": 0, "msg": "post success"}


api.add_namespace(hello_ns,"/hello")

if __name__ == '__main__':
    app.run(debug=True)
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

[2023-05-28 22:06:14] [INFO] [swagger_demo2.py]/[line: 54]/[post] request.args ===>ImmutableMultiDict([('choice', '2')]) 
127.0.0.1 - - [28/May/2023 22:06:14] "POST /hello?choice=2 HTTP/1.1" 200 -
1
2

十、orm介绍

1、什么是持久化

持久化（Persistence） 即把数据保存到可永久保存的存储设备中（如磁盘）。持久化的主要应用是将内存中的数据存储在关系型的数据库中。当然也可以存储在磁盘文件中、XML数据文件中等等。

2、什么是 ORM

Object Relational Mapping 对象关系映射
作用是在关系型数据库和对象之间做一个映射，这样在具体操作数据库的时候，就不需要再去和复杂的 SQL 语句打交道，只要像平时操作对象一样操作就可以了

3、ORM 与 SQL 的对比

# 使用sql语句查询数据
sql = "SELECT
username, email, gender
FROM User  WHERE id = 1"

# 执行sql 代码
res = db.execSql(sql);

# 取出数据
name = res[0]
1
2
3
4
5
6
7
8
9
10

# 使用orm 查询数据，定义ORM对象，get 值 ，
res = db.session.query(\
    User.username, User.email, User.gender).\
    filter(User.id==1).first()
name = res.username
1
2
3
4
5

4、ORM 优缺点

优势：

隐藏了数据访问细节
ORM 使我们构造固化数据结构变得非常简单

缺点：

性能下降 ，添加了关联操作，性能不可避免的会下降一些
无法解决特别复杂的数据库操作

十一、orm中间件配置

from flask import Flask
# 实例化 Flask的类，并且绑定module
from flask_sqlalchemy import SQLAlchemy
# 实例化 Flask
app=Flask(__name__)
# mysql 数据库用户名
username="root"
# mysql 数据库密码
pwd="root"
# mysql 数据库的 host 地址
ip="127.0.0.1"
# mysql 数据库端口
port="3306"
# 代码使用的数据库名
datebase="demo"
# 设置mysql 链接方法是
app.config['SQLALCHEMY_DATABASE_URL']=f"mysql+pymysql://{username}:{pwd}@{ip}:{port}/{database}?charset=utf-8"
# 定义应用使用数据库的配置
# 设置SQLALCHEMY_TRACK_MODIFICATIONS参数 不设置该配置的时候会抛出警告
app.config['SQLALCHEMY_TRACK_MODIFICATIONS']=True
# 将 app 与 Flask-SQLAlchemy 的 db 进行绑定
db=SQLAlchemy(app)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

十二、数据库与表管理

Flask-SQLAlchemy 数据库连接的配置

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@host:port/database'

database:
  username: root
  password: root
  server: 127.0.0.1:3306
  db: flask_1
1
2
3
4
5

定义数据库的表 需要继承 db.Model，db 为 app 启动的时的 SQLAlchemy 绑定的实例
类名，相当于表名
驼峰命名的类名，转为下划线连接。例如 class UserInfo 建立的表名为 user_info
自定义表名，__tablename__= 自定义的表名

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/5/28 22:37
# @Author  : 杜兰特
# @File    : sqlachemy_demo.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
# 实例化app 对象
from sqlalchemy import *
import yaml

app = Flask(__name__)
with open("./data.yaml") as f :
    result = yaml.safe_load(f)
    username = result.get("database").get('username')
    password = result.get("database").get('password')
    server = result.get("database").get('server')
    db = result.get("database").get('db')
app.config['SQLALCHEMY_DATABASE_URI'] = \
    f"mysql+pymysql://{username}:{password}@{server}/{db}?charset=utf8"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
# SQLAlchemy 绑定app
db = SQLAlchemy(app)

# 定义数据库的表 需要继承 db.Model，db 为 app 启动的时的 SQLAlchemy 绑定的实例

# 类名，相当于表名
#驼峰命名的类名，转为下划线连接。例如 class UserInfo 建立的表名为 user_info
#自定义表名，__tablename__= 自定义的表名
class User(db.Model):
    __tablename__="User"    #指定表名
    id = Column(Integer, primary_key=True)
    username = Column(String(80))

class StudentInfo(db.Model):
    id = Column(Integer, primary_key=True)
    username = Column(String(80))

if __name__ == '__main__' :
    #可以创建多个表
    db.create_all()

    #db.drop_all()
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

创建表操作

可以创建多个表
db.create_all()

删除表操作

db.drop_all()

十三、对象与数据模型

1、数据模型

数据模型（Data Model）是数据特征的抽象，它从抽象层次上描述了系统的静态特征、动态行为和约束条件，为数据库系统的信息表示与操作提供一个抽象的框架。

2、设计用户表字段

3、Flask-SQLAlchemy 的属性字段定义

通常类的属性相当于表的一个字段
定义的属性的方式为 name=Column(参数的类型, 其他的属性)
官方：https://flask-sqlalchemy.palletsprojects.com/en/2.x/models/?highlight=column

4、Flask-SQLAlchemy 字段常用关键字参数

5、Flask-SQLAlchemy 对象与数据模型示例

每一个类变量表示一个数据库表的列名
第一个参数是表示数据的类型， primary_key=True 表示是主键
unique = True 表示是不能重复的
nullable=False 表示不可以为空

from sqlalchemy import *
from flask_sqlalchemy import SQLAlchemy
import yaml
# 导入 Query 以便于调用的时候代码提示
from sqlalchemy.orm import Query
from flask import Flask
app = Flask(__name__)

with open("./data.yaml") as f :
    result = yaml.safe_load(f)
    username = result.get("database").get('username')
    password = result.get("database").get('password')
    server = result.get("database").get('server')
    db = result.get("database").get('db')
app.config['SQLALCHEMY_DATABASE_URI'] = \
    f"mysql+pymysql://{username}:{password}@{server}/{db}?charset=utf8"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
# SQLAlchemy 绑定app
db = SQLAlchemy(app)

# 定义数据库的表 需要继承 db.Model
class User(db.Model):
    __tablename__ = "user" # 设置数据库表名
    # 每一个类变量表示一个数据库表的列名
    # 第一个参数是表示数据的类型， primary_key=True 表示是主键
    id = Column(Integer, primary_key=True)
    # unique = True 表示是不能重复的  nullable=False 表示不可以为空
    username = Column(String(80), unique=False, nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    gender = Column(String(3),unique=False)

    def __repr__(self):
        # 数据库的 魔法方法 直观展示数据
        '''[,]'''
        return f'{self.username}>'
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

十四、数据CRUD

1、添加数据（create）

a、单条数据新增

实例化类，创建表数据
将实例添加到 session（add）
提交更新 (commit)
关闭 session

# 新增表数据，需要导入 User 类，进行实例化
user=User(id=1,username="kobe",email="111@qq.com",gender="男")
print(user)
db.session.add(user)
db.session.commit()
db.session.close()
1
2
3
4
5
6

b、多条数据新增

多次实例化类，创建多条表数据
将多个实例依次添加到 session 中（add）或者一次性添加到 session 中（add_all）
提交更新 (commit)
关闭 session

# 批量添加数据操作
user1 = User(username="kobe1", email="1113@qq.com", gender="男")
user2 = User(username="kobe2", email="1114@qq.com", gender="女")
db.session.add_all([user1,user2])
db.session.commit()
db.session.close()
1
2
3
4
5
6

2、读取数据（read）

a、查询表中全部数据

格式：类.query.all()

# 读取全部数据
res = User.query.all()
# 遍历数据，得到想要的字段
for rs in res:
    print(rs.username, rs.email)
1
2
3
4
5

b、条件查询——单条件查询

格式：类.query.filter_by(条件).单条或多条

查询单条数据：类.query.filter_by(条件).first()

res=User.query.filter_by(gender='男').first()
print(res)
1
2

查询多条数据：类.query.filter_by(条件).all()

res_list=User.query.filter_by(gender="男").all()
print(res_list)
1
2

c、条件查询——多条件查询

查询单条数据：类.query.filter_by(条件).filter_by(条件)…first()

res1=User.query.filter_by(gender="男").filter_by(username="kobe1").first()
print(res1)
1
2

查询多条数据：类.query.filter_by(条件).filter_by(条件)…all()

res2 = User.query.filter_by(gender="男").filter_by(username="kobe1").all()
print(res2)
1
2

3、修改数据（update）

方式一：
首先查询出来需要的数据
对查询出来的数据对象进行属性的修改
提交 session

user3=User.query.filter_by(id=2).first()
user3.gender="女"
db.session.commit()
db.session.close()
1
2
3
4

方式二：
给定查询条件进行查询后，直接进行 update 操作
提交 session

res4=User.query.filter_by(id=3).update({"username":"butler"})
db.session.commit()
db.session.close()
1
2
3

4、删除数据（delete）

方式一：
查询数据
对查询出来的数据对象进行删除操作
提交 session

user5=User.query.filter_by(id=1).first()
db.session.delete(user5)
db.session.commit()
db.session.close()
1
2
3
4

方式二：
给定查询条件进行查询后，直接进行 delete 操作
提交 session

user5=User.query.filter_by(id=2).delete()
db.session.commit()
db.session.close()
1
2
3

十五、多表关系-一对多

1、一对多关系

场景：
一个班级有多个学生
一个学生只在一个班级
一对多关系中,通过外键来关联数据，外键设置在多的一方

SQLAlchemy提供了一个relationship，这个类可以定义属性，以后在访问相关联的表的时候就直接可以通过属性访问的方式就可以访问得到了。另外，可以通过backref来指定反向访问的属性名称。

特别注意：
classinfo=db.relationship(‘ClassInfo’,backref=“studentinfo”)：代表学生所属的班级信息，backref:反向关联的属性名，classinfo=db.relationship(‘ClassInfo’,backref=“studentinfo”)：中的ClassInfo必须是另外一个相关联的类名

from sqlalchemy import *
from flask_sqlalchemy import SQLAlchemy
import yaml
# 导入 Query 以便于调用的时候代码提示
from sqlalchemy.orm import Query
from flask import Flask
app = Flask(__name__)

with open("../data.yaml") as f :
    result = yaml.safe_load(f)
    username = result.get("database").get('username')
    password = result.get("database").get('password')
    server = result.get("database").get('server')
    db = result.get("database").get('db')
app.config['SQLALCHEMY_DATABASE_URI'] = \
    f"mysql+pymysql://{username}:{password}@{server}/{db}?charset=utf8"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
# SQLAlchemy 绑定app
db = SQLAlchemy(app)


# 班级表
class ClassInfo(db.Model):
    __tablename__ = "Class"
    id = Column(Integer, primary_key=True)
    name = Column(String(80))

    def __repr__(self):
        return f"{self.name}>"

# 学生表
class StudentInfo(db.Model):
    # 指定表名__tablename__属性
    __tablename__ = "Student"
    id = Column(Integer, primary_key=True)
    name = Column(String(80))
    #添加外键
    classid = Column(Integer, ForeignKey("Class.id"))

    # 第一个参数：表示这个关系的另一端是 ClassInfo 类。比如班级id=1
    # 第二个参数：backref, 表示反向引用，需要从班级id为1 反向获取【多】的时候，使用的属性
    # 通过ClassInfo实例.studentinfo
    classinfo=db.relationship('ClassInfo',backref="studentinfo")

    def __repr__(self):
        return f"{self.name}>"
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

2、一对多表结构

创建表
db.create_all()
删除表
db.drop_all()

3、一对多增删查改

a、数据新增

Class 班级表：添加两条数据

class1 = ClassInfo(id=1, name="测开21期")
class2 = ClassInfo(id=2, name="测开22期")
db.session.add_all([class1, class2])
db.session.commit()
db.session.close()
1
2
3
4
5

要先完全创建好，数据库中的班级表，再去创建学生表
Student 学生表：添加两条数据

student1 = StudentInfo(id=1, name="学生一", classid=1)
student2 = StudentInfo(id=2, name="学生二", classid=1)
student3 = StudentInfo(id=3, name="学生三", classid=2)
student4 = StudentInfo(id=4, name="学生四", classid=2)
db.session.add_all([student1, student2,student3, student4])
db.session.commit()
db.session.close()
1
2
3
4
5
6
7

b、数据查询

多查一

通过学生id=1 查询学生所在的班级

stu1=StudentInfo.query.filter_by(id=1).first()
#获取classid信息
print(stu1.classid)
print(stu1.classinfo)
print(stu1.classinfo.name)
1
2
3
4
5

一查多

通过班级id=1，查询对应的学生

class1=ClassInfo.query.filter_by(id=1).first()
print(class1.studentinfo)
print(class1.studentinfo[0].name)
1
2
3

c、数据修改

一改多

通过班级修改学生信息

class1=ClassInfo.query.filter_by(id=1).first()
print(class1.studentinfo)
class1.studentinfo[0].name="学生修改1"
db.session.commit()
db.session.close()
1
2
3
4
5

多改一

通过学生修改班级信息

stu2=StudentInfo.query.filter_by(id=2).first()
print(stu2.classinfo)
stu2.classinfo.name="pythonvip"
db.session.commit()
db.session.close()
1
2
3
4
5

d、数据删除

删除一个班级下的所有学生

class1=ClassInfo.query.filter_by(id=1).first()
StudentInfo.query.filter(StudentInfo.classid==class1.id).delete()
db.session.commit()
db.session.close()
1
2
3
4