Contents
23.6.3. Flask¶
1.Flask综述¶
相对于其他Python语言的Web框架而言，Flask的特点可以归结如下。
1．内置开发服务器和调试器
2．与Python单元测试功能无缝衔接
3．使用Jinja2模板
4．完全兼容WSGI 1.0标准
5．基于Unicode编码
2.实战演练:开发Flask API视图¶
2.1 Hello World程序¶
现在可以开始写第1个Flask程序了，一个最简单的Flask程序应该包括如下内容：
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world(): # put application's code here
return 'Hello World!'
if __name__ == '__main__':
app.run()
当确定系统可以接受来自外部的访问时，可以通过给run（）方法设置参数的方式来实现。同时，在run参数中可以传入要监听的端口，并且设置是否处于调试模式运行。如果将helloworld.py文件中的app.run()改为如下命令，则系统将会监听所有地址的80端口，并关闭调试模式：
app.run(host='0.0.0.0', port=5000, debug=True)
2.2 模板渲染¶
1.用render_template实现模板渲染¶
使用模板渲染的代码如下：
from flask import Flask
from flask import render_template
app = Flask(__name__)
@app.route('/')
def hello_world(): # put application's code here
return 'Hello World!'
@app.route('/hello')
@app.route('/hello/<name>')
def hello(name=None):
return render_template('hello.html', name=name)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
假设本例的Python代码被保存在app.py中，则本例的网站目录结构如下：
/app.py
/templates
/hello.html
下面是hello.html文件的内容：
<! doctype html>
<title>Hello from Flask</title>
{% if name %} <! -判断name参数是否为空 →
<h1>Hello {{ name }}! </h1> <! -当name参数不为空时，本行生效 →
{% else %}
<h1>Hello World! </h1> <! -当name参数为空时，本行生效 →
{% endif %}
2.用Markup转换变量中的特殊字符¶
向render_template传入的参数，不仅可以是单纯的字符串，还可以包含HTML特殊字符（比如<、>、空格、/等），这给模板参数提供了更好的灵活性。同时，因为这些特殊字符会被HTML客户端解释成特殊含义，所以会给网站程序带来一定程度的安全隐患。Flask允许程序员自己控制Jinja2是否需要解释这些特殊字符。
如果这些字符应该被解释成特殊含义，则将这些参数直接传给render_template即可；如果这些字符仅应该被解释成字符串，则应该通过Markup()函数将这些字符串做转义处理，然后传给render_template()函数。
@app.route('/hello')
@app.route('/hello/<name>')
def hello(name=None):
from flask import Markup
print(Markup('<strong>Hi %s! </strong>') % '<blink>David</blink>')
return render_template('hello.html', name=name)
这段代码显示在浏览器中将会是一段粗体的“Hi David!”。如果不进行Markup转义，则将会在浏览器上显示成闪烁着的粗体字“Hi David!”。
3.重定向和错误处理¶
重定向（Redirect）是指将一个网络请求重新指定URL并转到其他地址的技术。Flask的redirect()函数提供了这个功能。此外，如果仅仅想中止一个请求并返回错误，而不是重定向到其他地址，则可以使用abort()函数。
例如：
from flask import Flask
from flask import abort, redirect
app = Flask(__name__)
@app.route('/')
def index():
return redirect('/check') # 重定向到/login页面
@app.route('/check')
def f_check():
abort(400) # 立即向客户端返回400错误
# dont_coding_here() #这里的代码不会被执行
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
本例中，当客户端访问根页面时，处理函数index()通过redirect()函数将请求重定向到了check页面。而check页面中目前没有实现其他逻辑，仅仅向客户端返回了400错误。
说明： 400是一个HTTP的标准错误定义，其含义为请求无效。
HTTP 定义了标准的返回码错误代码表，其中大于等于400的代码被认为错误。
客户端的浏览器遇到错误返回时会显示默认的错误页面。如果网站程序需要定义自己的错误页面，则可以通过添加错误处理器来实现。
例如可以在程序中添加如下代码段：
from flask import Flask
from flask import abort, redirect
from flask import render_template
app = Flask(__name__)
@app.route('/')
def index():
return redirect('/check') # 重定向到/login页面
@app.route('/check')
def f_check():
abort(400) # 立即向客户端返回400错误
# dont_coding_here() #这里的代码不会被执行
@app.errorhandler(400)
def bad_request(error):
return render_template('bad_request.html'), 400
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
本例中用Flask的errorhandler装饰器添加了自定义的错误处理器。当程序中返回400错误时，系统会执行bad_request()函数。在该函数中向客户端返回400错误的同时，还传送了自定义的错误页面bad_request.html。
2.3 Flask视图高级技术¶
视图绑定¶
使用@app.route装饰器
@app.route('/',endpoint='index')
#底层其实是使用add_url_rule实现的
def hello_world():
return 'Hello World!'
使用add_url_rule来绑定视图函数
def my_test():
return '这是测试页面!'
app.add_url_rule(rule='/test/',endpoint='test',view_func=my_test)
with app.test_request_context():
print(url_for('test'))
FBV与CBV¶
FBV¶
FBV的定义、路由映射方法、装饰器添加方法，Flask中通常会用FBV；
Flask框架中不常用CBV；更多的是用FBV
from flask import Flask,views
app = Flask(__name__)
def wrapper(func):
def inner(*args,**kwargs):
print('beforeFunc')
ret = func(*args,**kwargs)
return ret
return inner
# FBV
@app.route('/index1',methods=['GET','POST'])
@wrapper
def index1():
return "index1"
if __name__ == '__main__':
app.run()
CBV¶
标准类视图
标准类视图的特点：
必须继承flask.views.View。
必须实现dispatch_request方法，以后请求过来后，都会执行这个方法，这个方法的返回值相当于之前的视图函数，也必须返回Response或者子类的对象，或者是字符串、元祖。
必须通过app.add_url_rule(rule,endpoint,view_func)来做URL与视图的映射，view_func参数需要使用as_view类方法转换。
如果指定了endpoint，那么在使用url_for反转时就必须使用endpoint指定的那个值。如果没有指定endpoint，那么就可以使用as_view(视图名称)中指定的视图名称来作为反转。
如果一个网站有首页、注册页和登录页面，每个页面要求放置一个同样的对联广告，使用类视图函数如何实现呢？
from flask import Flask,render_template,views
app = Flask(__name__)
class Ads(views.View):
def __init__(self):
super().__init__()
self.context = {
'ads': '这是对联广告！'
}
class Index(Ads):
def dispatch_request(self):
return render_template('index.html', **self.context)
class Login(Ads):
def dispatch_request(self):
return render_template('login.html', **self.context)
class Register(Ads):
def dispatch_request(self):
return render_template('register.html', **self.context)
app.add_url_rule(rule='/', endpoint='index', view_func=Index.as_view('Index'))
app.add_url_rule(rule='/login/', endpoint='login', view_func=Login.as_view('login'))
app.add_url_rule(rule='/register/', endpoint='register', view_func=Register.as_view('register'))
if __name__ == '__main__':
app.run(debug=True)
基于方法的类视图
利用视图函数实现不同的请求执行不同的逻辑时比较复杂，需要在视图函数中进行判断，如果利用方法视图实现就比较简单。
Flask提供了另外一种类视图flask.views.MethodView，
对每个HTTP方法执行不同的函数（映射到对应方法的小写的同名方法上）。
在CBV中，路由信息只能通过add_url_rule()方法添加
from flask import Flask, views
app = Flask(__name__)
class Login(views.MethodView):
methods = ['POST', 'GET']
# 如果需要在CBV中加装饰器的话,括号里就是装饰器的内存地址，可以传多个
decorators = ()
def get(self):
print('get 请求')
return 'login get'
def post(self):
print('post 请求')
return 'login post'
app.add_url_rule('/login', view_func=Login.as_view(name='login'))
if __name__ == "__main__":
app.run(debug=True)
基于方法视图的用户登录
from flask import Flask,render_template,request,views#导入相应模块
app = Flask(__name__)#Flask初始化
@app.route('/')#定义路由
def hello_world():#定义视图函数
return render_template('index.html')#渲染模板
class LoginView(views.MethodView):#定义LoginView类
# 当用户通过get方法进行访问的时候执行get方法
def get(self):#定义get函数
return render_template("index.html")#渲染模板
# 当用户通过post方法进行访问的时候执行post方法
def post(self):#定义post 函数
username = request.form.get("username")#接收表单中传递过来的用户名
password = request.form.get("pwd")#接收表单中传递过来的密码
if username == 'admin' and password == 'admin':#如果用户名和密码是否为admin
return "用户名正确，可以登录！"#i f语句为真的话，返回可以登录信息
else:
return "用户名或密码错误,不可以登录！"#否则，返回不可以登录信息
# 通过add_url_rule添加类视图和url的映射关系
app.add_url_rule('/login',view_func=LoginView.as_view('loginview'))
if __name__ == '__main__': #当模块被直接运行时，代码将被运行，当模块是被导入时，代码不被执行
app.run(debug=True)#开启调试模式
2.4 Flask装饰器¶
装饰器本质上是一个Python函数，它可以让其他函数在不需要做任何代码变动的前提下增加额外的功能，装饰器的返回值也是一个函数对象。
装饰器经常用于有切面需求的场景，
比如插入日志、性能测试、事务处理、缓存和权限校验等场景。
装饰器是解决这类问题的绝佳设计，有了装饰器，可以抽离出大量与函数功能无关的雷同代码并继续重用。
1.装饰器的定义和基本使用¶
在视图函数news()之前直接加@user_login，就实现了装饰器的使用。
from flask import Flask # 导入Flask模块
app = Flask(__name__) # Flask初始化
@app.route('/') # 定义路由
def hello_world(): # 定义函数
return 'Hello World!' # 返回值
def user_login(func): # 定义函数，使用func接收函数作为参数
def inner(): # 定义inner()函数
print('登录操作') # 打印输出
func() # 执行func函数
return inner # 返回inner函数，不是返回函数的结果
@user_login # 使用了装饰器
def news(): # 定义函数news
print('这是新闻详情页') # 打印输出
news() # 调用news来执行
if __name__ == '__main__':
app.run()
2.对带参数的函数使用装饰器¶
1.函数的可变参数
def func(*args,**kwargs):
- *：代指元组，长度不限；
- **：代表键值对，个数不限。
- *args：代表元祖，长度不限制。
- **kwargs：代表按键值对，个数不限。
2.对带参数的函数使用装饰器
from flask import Flask # 导入Flask模块
app = Flask(__name__) # Flask初始化
@app.route('/') # 定义路由
def hello_world(): # 定义视图函数
return 'Hello World!' # 返回值
def user_login(func): # 定义函数user_login
def inner(*args, **kwargs): # 定义内部函数inner
print('登录操作!') # 打印输出
func(*args, **kwargs)
return inner # 返回inner
@user_login # 使用装饰器
def news(): # 定义函数news()
print(news.__name__) # 打印输出此时的函数名称
print('这是新闻详情页!') # 打印输出
news()
@user_login # 使用装饰器
def news_list(*args): # 定义函数news_list
page = args[0] # 元祖args[0]赋值给page
print(news_list.__name__) # 打印输出函数名
print('这是新闻列表页的第'+str(page)+'页！') # 打印输出
news_list(5) # 调用函数news_list
if __name__ == '__main__': # 当模块被直接运行时，代码将被运行，当模块是被导入时，代码不被执行
app.run()
输出结果如下：
登录操作!
inner
这是新闻详情页!
登录操作!
inner
这是新闻列表页的第5页！
以上代码还存在一点问题，在调用过程中会改变原来的名称，不管是news()函数还是news_list()函数，最终执行时被替换成了inner()函数。
为避免出现此种情况，可以使用functools.wraps在装饰器的函数上对传进来的函数进行包裹，这样就不会丢失原始函数了。
注意：导入包wraps，使用命令为from functools import wraps。
带参数的函数使用装饰器优化实例：app.py
from flask import Flask # 导入Flask模块
from functools import wraps # 导入相应模块
app = Flask(__name__) # Flask初始化
@app.route('/') # 定义路由
def hello_world(): # 定义视图函数
return 'Hello World!' # 返回值
def user_login(func): # 定义函数user_login
@wraps(func) # 使用functools.wraps在装饰器的函数上，把传进来的函数进行包裹
def inner(*args, **kwargs): # 定义内部函数inner
print('登录操作!') # 打印输出
func(*args, **kwargs)
return inner # 返回inner
@user_login # 使用装饰器
def news(): # 定义函数news()
print(news.__name__) # 打印输出此时的函数名称
print('这是新闻详情页!') # 打印输出
news()
@user_login # 使用装饰器
def news_list(*args): # 定义函数news_list
page = args[0] # 元祖args[0]赋值给page
print(news_list.__name__) # 打印输出函数名
print('这是新闻列表页的第'+str(page)+'页！') # 打印输出
news_list(5) # 调用函数news_list
if __name__ == '__main__': # 当模块被直接运行时，代码将被运行，当模块是被导入时，代码不被执行
app.run()
3.路由详解¶
3.1在路径中添加变量¶
@app.route('/login/<username>')
def show_welcome(username):
return f'Hi {username}'
被添加的参数需要两次被声明：第1次是在route()装饰器的参数中，在需要使用变量的URL部分用方式声明变量；第2次是在所映射的函数（本例中为show_welcome）的参数中声明变量名，这样被声明的变量就可以在映射函数内使用了。
本例中username变量被作为欢迎语句的一部分回传给客户端。
注意： 两次变量声明的变量名一定要一致。
3.2为变量指定类型¶
可以在声明变量时指定变量被映射的类型，比如：
@app.route('/add/<int:number>')
def add_one(number):
return f"{number + 1}"
Flask中允许有3种类型的变量映射，如表8.1所示。
表8.1 路由变量映射类型表
映射类型 |
说明 |
|---|---|
Int |
接受整型数值变量 |
Float |
接受浮点型数值变量 |
Path |
默认方式，接受路径字符串 |
3．路径最后的分隔符的作用
在URL路径中，斜杠“/”被用作路径分隔符。当斜杠被写在URL路径的开头时，则表明本路径是一个绝对路径；当斜杠被写在路径中间时，它被用作隔离路径的层级。
那么，当它被写在最后时，它的作用是什么呢？
通过下面的例子可以理解斜杠分隔符被写在路径最后时的作用。
@app.route('/school/')
def schools():
return 'The school page'
@app.route('/school')
def schools():
return 'The school page'
是否带斜杠结尾的路径声明效果分析
3.3HTTP方法绑定¶
网站通过HTTP与浏览器或其他客户端进行交互，而HTTP访问一个URL时可以使用几种不同的访问方式，包括GET、POST、HEAD、DELETE等。在Flask中，路由默认设置使用GET方式进行路径访问。
1.指定HTTP访问方式的方法
通过修改route中的参数，可以配置其他访问方式。举例说明：
@app.route('/SendMessage', methods=['GET', 'POST'])
def Messaging():
if request.method == 'POST':
do_send()
else:
show_the_send_form()
本例中，在route装饰器中显式地声明了两种HTTP访问方式：GET和POST。无论客户端使用哪种方式访问地址/SendMessage, Flask都会定位到Messaging（）函数并执行。可以在函数中通过request.method属性获得本次HTTP请求的访问方式。
注意： request是Flask框架的一个全局对象，可以获得很多HTTP请求的客户端相关的信息。
2.将同一个URL根据访问方式映射到不同的函数
可以灵活地运用URL、访问方法、被映射函数的绑定关系。通过把不同的访问方式赋予相同的URL，可以对其绑定不同的映射函数。
【示例8-8】比如：
@app.route('/Message', methods=[ 'POST'])
def do_send():
return "This is for POST methods"
@app.route('/Message', methods=['GET'])
def show_the_send_form():
return "This is for GET methods"
3.4路由地址反向生成¶
通过前面的学习，读者已经掌握了将URL绑定到映射函数的方法。但有时，程序中需要通过函数名称获得与其绑定的URL地址。Flask通过url_for（）函数实现了这个功能。在使用前，需要先从Flask包中导入对url_for（）函数的引用。函数的第1个参数是需要获取URL的函数名，URL中如果有变量，则可以在url_for（）中添加参数来实现对变量的赋值。
from flask import Flask, url_for
app = Flask(__name__)
@app.route('/')
def f_root():pass
@app.route('/industry')
def f_industry():pass
@app.route('/book/<book_name>')
def f_book(book_name):pass
with app.test_request_context():
print(url_for('f_root')) #例1，输出：/
print(url_for('f_industry')) #例2，输出：/industry
print(url_for('f_industry', name='web')) #例3，输出：/industry? name=web
print(url_for('f_book', book_name='Python Book')) #例4，输出：/book/Python%20Book
技巧： Flask中test_request_context（）方法用于告诉解释器为在其作用域中的代码模拟一个HTTP请求上下文，使其好像被一个HTTP请求所调用。HTTP请求上下文是调用url_for所必需的环境。
我们可以给这个装饰器再加上endpoint参数（给这个URL命名）
@app.route('/',endpoint='index')
def hello_world():
return 'hello world'
一旦我们使用了endpoint参数，在使用url_for()反转时就不能使用视图函数名了，而是要用我们定义的URL名。
url_for('index')
在程序中需要使用url_for()函数的原因如下。
反向解析比硬编码有更好的可读性和可维护性。比如，当需要更换路由函数中URL的地址时，无须再更改和调用url_for处的代码。
url_for会自动处理必需的特殊字符转换和Unicode编码转换。本节代码段中例4的空格就被自动解析为%20。
3.5路由系统¶
@app.route(‘/user/’)
@app.route(‘/post/’)
@app.route(‘/post/’)
@app.route(‘/post/’)
@app.route(‘/login’, methods=[‘GET’, ‘POST’])
常用路由系统有以上五种，所有的路由系统都是基于一下对应关系来处理：
DEFAULT_CONVERTERS = {
'default': UnicodeConverter,
'string': UnicodeConverter,
'any': AnyConverter,
'path': PathConverter,
'int': IntegerConverter,
'float': FloatConverter,
'uuid': UUIDConverter,
}
4.静态文件¶
静态文件，顾名思义，就是那些不会被改变的文件，比如图片，CSS 文件和 JavaScript 源码文件。默认情况下，Flask 在程序根目录中名为 static 的子目录中寻找静态文件。因此，我们一般在应用的包中创建一个叫 static 的文件夹，并在里面放置我们的静态文件。比如，我们可以按下面的结构组织我们的 app：
app/
__init__.py
static/
css/
style.css
home.css
admin.css
js/
home.js
admin.js
img/
favicon.co
logo.svg
templates/
index.html
home.html
admin.html
views/
models/
run.py
但是，我们有时还会应用到第三方库，比如 jQuery, Bootstrap 等，这时我们为了不跟自己的 Javascript 和 CSS 文件混起来，我们可以将这些第三方库放到 lib 文件夹或者 vendor 文件夹，比如下面这种：
static/
css/
lib/
bootstrap.css
style.css
home.css
admin.css
js/
lib/
jquery.js
chart.js
home.js
admin.js
img/
logo.svg
favicon.ico
1.提供一个favicon 图标¶
favicon 是 favorites icon 的缩写，也被称为 website icon（网页图标）、page icon（页面图标）等。通常而言，定义一个 favicon 的方法是将一个名为『favicon.ico』的文件置于 Web 服务器的根目录下。但是，正如我们在上面指出，我们一般将图片等静态资源放在一个单独的 static 文件夹中。为了解决这种不一致，我们可以在站点模板的 部分添加两个 link 组件，比如我们可以在 template/base.html 中定义 favicon 图标：
{% block head %}
{{ super() }}
<link rel="shortcut icon" href="{{ url_for('static', filename = 'favicon.ico') }}" type="image/x-icon">
<link rel="icon" href="{{ url_for('static', filename = 'favicon.ico') }}" type="image/x-icon">
{% endblock %}
在上面的代码中，我们使用了 super()
来保留基模板中定义的块的原始内容，并添加了两个 link
组件声明图标位置，这两个 link 组件声明会插入到 head 块的末尾。
5. Jinja2 模板引擎¶
了解即可，不用深入
目前flask主要用来写接口前后端分离，Jinja2多用于前后端不分离场景，暂时接触不多，部分可以阅读如下文章
https://www.bookstack.cn/read/head-first-flask/chapter02-section2.04.md
6.请求、重定向及会话¶
Web 开发中经常需要处理 HTTP 请求、重定向和会话等诸多事务，相应地，Flask 也内建了一些常见的对象如 request, session, redirect 等对它们进行处理。
1.请求对象 request¶
HTTP 请求方法有 GET、POST、PUT 等，request 对象也相应地提供了支持。
举个例子，假设现在我们开发一个功能：用户注册。如果 HTTP 请求方法是 POST，我们就注册该用户，如果是 GET 请求，我们就显示注册的字样。代码示例如下（注意，下面代码并不能直接运行，文末提供了完整的代码）：
from flask import Flask, request
app = Flask(__name__)
@app.route('/register', methods=['POST', 'GET']):
def register():
if request.method == 'GET':
return 'please register!'
elif request.method == 'POST':
user = request.form['user']
return 'hello', user
2.重定向对象 redirect¶
当用户访问某些网页时，如果他还没登录，我们往往会把网页重定向到登录页面，Flask 提供了 redirect 对象对其进行处理，我们对上面的代码做一点简单的改造，如果用户注册了，我们将网页重定向到首页。代码示例如下：
from flask import Flask, request, redirect
app = Flask(__name__)
@app.route('/home', methods=['GET']):
def index():
return 'hello world!'
@app.route('/register', methods=['POST', 'GET']):
def register():
if request.method == 'GET':
return 'please register!'
elif request.method == 'POST':
user = request.form['user']
return redirect('/home')
3.会话对象 session¶
程序可以把数据存储在用户会话中，用户会话是一种私有存储，默认情况下，它会保存在客户端 cookie 中。Flask 提供了 session 对象来操作用户会话，下面看一个示例：
from flask import Flask, request, session, redirect, url_for, render_template
app = Flask(__name__)
@app.route('/home', methods=['GET'])
def index():
return 'hello world!'
@app.route('/register', methods=['GET', 'POST'])
def register():
if request.method == 'POST':
user_name = request.form['user']
session['user'] = user_name
return 'hello, ' + session['user']
elif request.method == 'GET':
if 'user' in session:
return redirect(url_for('index'))
else:
return render_template('login.html')
app.secret_key = '123456'
if __name__ == '__main__':
app.run(host='127.0.0.1', port=5632, debug=True)
操作 session 就像操作 python 中的字典一样，我们可以使用
session['user'] 获取值，也可以使用 session.get('user')
获取值。注意到，我们使用了 url_for 生成 URL，比如 /home 写成了
url_for('index')。
url_for()
函数的第一个且唯一必须指定的参数是端点名，即路由的内部名字。默认情况下，路由的端点是相应视图函数的名字，因此
/home 应该写成 url_for('index')。
还有一点，使用session 时要设置一个密钥 app.secret_key。
4.附录¶
本节完整的代码如下：
$ tree .
.
├── flask-session.py
└── templates
├── layout.html
└── login.html
$ cat flask-session.py
from flask import Flask, request, session, redirect, url_for, render_template
app = Flask(__name__)
@app.route('/')
def head():
return redirect(url_for('register'))
@app.route('/home', methods=['GET'])
def index():
return 'hello world!'
@app.route('/register', methods=['GET', 'POST'])
def register():
if request.method == 'POST':
user_name = request.form['user']
session['user'] = user_name
return 'hello, ' + session['user']
elif request.method == 'GET':
if 'user' in session:
return redirect(url_for('index'))
else:
return render_template('login.html')
app.secret_key = '123456'
if __name__ == '__main__':
app.run(host='127.0.0.1', port=5632, debug=True)
$ cat layout.html
<!doctype html>
<title>Hello Sample</title>
<link rel="stylesheet" type="text/css" href="{{ url_for('static', filename='style.css') }}">
<div class="page">
{% block body %}
{% endblock %}
</div>
$ cat login.html
{% extends "layout.html" %}
{% block body %}
<form name="register" action="{{ url_for('register') }}" method="post">
Hello {{ title }}, please login by:
<input type="text" name="user" />
</form>
{% endblock %}
7.蓝图¶
一个程序执行文件中，如果功能代码过多，是不方便后期维护的。如何实现程序代码模块化，根据具体不同功能模块的实现，划分成不同的分类，降低各功能模块之间的耦合度呢？这时flask.Blueprint（蓝图）就派上用场了。
注意：蓝图可以极大地简化大型应用，并为扩展提供集中的注册入口。
Blueprint对象与Flask应用对象的工作方式类似，但不是一个真正的应用。
在PyCharm中新建工程，新建app.py、news.py、produt.py三个文件。
app.py文件内容如下：
蓝图使用实例：主路由视图函数app.py
# -*- coding:utf-8 -*-
from flask import Flask # 导入Flask模块
import news, products # 导入相应模块
app = Flask(__name__) # 创建 Flask()对象： app
@app.route('/') # 使用了蓝图，app.route() 这种模式就仍可以使用，注意路由重复的问题
def hello_world(): # 定义函数
return 'hello my world !' # 返回值
app.register_blueprint(news.new_list) # 将news模块里的蓝图对象new_list注册到app
app.register_blueprint(products.product_list) # 将products模块里的蓝图对象product_list注册到app
if __name__ == '__main__':
app.run(debug=True) # 调试模式开 启动服务器 运行在默认的5000端口
news.py文件内容如下：
蓝图使用实例：分路由视图函数news.py
# -*- coding:utf-8 -*-
from flask import Blueprint # 导入Blueprint模块
new_list = Blueprint('news', __name__) # 创建一个blueprint对象。第一个参数可看做该blueprint对象的姓名
# 在一个app里，姓名不能与其余的Blueprint对象姓名重复
# 第二个参数__name__用作初始化
@new_list.route("/news") # 将蓝图对象当做‘app’那样使用
def new(): # 定义函数news()
return '这是新闻模块！'
products.py文件内容如下：
蓝图使用实例：分路由视图函数products.py
# -*- coding:utf-8 -*-
from flask import Blueprint # 导入Blueprint模块
product_list = Blueprint('products', __name__) # 创建一个blueprint对象。第一个参数可看做该blueprint对象的名字
# 在一个app里，对象名不能与其余的Blueprint对象名重复
# 第二个参数__name__用作初始化
@product_list.route("/products") # 将蓝图对象当做‘app’那样使用
def product():
return '这是产品模块！'
查看蓝图设置是否生效:
$ curl http://127.0.0.1:5000/
hello my world !
$ curl http://127.0.0.1:5000/news
这是新闻模块！
$ curl http://127.0.0.1:5000/products
这是产品模块！
蓝图的目的是实现各个模块的视图函数写在不同的py文件中。
在主视图中导入分路由视图的模块，并且注册蓝图对象。
注意：视图函数的名称不能和蓝图对称的名称一样。
https://www.bookstack.cn/read/head-first-flask/chapter02-section2.06.md
8.Flask数据交互¶
8.1 使用Flask处理表单¶
Form表单是Web应用中最基础的一部分。为了能处理Form表单，Flask-WTF扩展提供了良好的支持。
1.使用Flask处理通用表单¶
Flask请求对象包含客户端发出的所有请求信息。其中，request.form能获取POST请求中提交的表单数据。
尽管Flask的请求对象提供的信息足够用于处理Web表单，但有些任务很单调，而且需要重复操作。
使用Flask处理普通表单：app.py
# encoding:utf-8#指定编码
from flask import Flask, render_template, request # 导入相应模块
app = Flask(__name__) # Flask初始化
@app.route('/') # 定义路由
def hello_world(): # 定义视图函数
return render_template('index.html') # 使用render_template函数渲染模板
@app.route('/login', methods=['GET', 'POST']) # 定义路由，指定访问方法
def login(): # 定义视图函数
if request.method == 'GET': # 如果访问方法为GET方法
return '这是get请求' # 返回应答信息
else:
return '这是POST请求' # 返回应答信息
if __name__ == '__main__': # 当模块被直接运行时，代码将被运行，当模块是被导入时，代码不被运行。
app.run(debug=True) # 开启调试模式
使用Flask处理普通表单：index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8"><!--指定网页编码-->
<title>Title</title><!--指定网页标题-->
<style type="text/css">
.div1 {
height:180px;
width:380px;
border:1px solid #8A8989;
margin:0 auto;
}
.input{
display: block;
width: 350px;
height: 40px;
margin: 10px auto;
}
.button
{
background: #2066C5;
color: white;
font-size: 18px;
font-weight: bold;
height: 50px;
border-radius: 4px;
}
</style>
</head><!--head区域完毕-->
<body><!--body区域开始-->
<div class="div1"><form action="login" method = "post"><!--定义表单开始-->
<input type="text" class="input" placeholder="请输入用户名"><!--输入文本框-->
<input type="password" class="input" placeholder="请输入密码"><!--输入文本框-->
<input type="submit" value="登录" class="input button"><!--定义登陆用button-->
</form></div> <!--form表单结束-->
</body><!--body区域完毕-->
</html>
在上面的工程中，对表单没有进行必要的保护措施，很容易被人利用，控制用户在当前已登录的Web应用程序上执行非本意的操作。
因此，在实际部署服务器上的代码时，不建议使用这个方式处理表单，推荐使用Flask-WTF方式进行表单处理。
2.使用Flask-WTF处理表单¶
1．Flask-WIF的安装
$ pip install flask-wtf
2．启用CSRF保护
Flask-WTF提供了对所有Form表单免受跨站请求伪造（Cross-Site Request Forgery，CSRF）攻击的技术支持（通过添加动态token令牌的方式）。
我们在Flask根目录下新增config.py配置文件，要启用CSRF保护，可以在config.py中定义两个变量：
CSRF_ENABLED= True
SECRET_KEY= 'x1x2x3x4x5x6'
其中，SECRET_KEY用来建立加密的令牌，用于验证Form表单提交，在自己编写应用程序时，可以尽可能设置得复杂一些，这样恶意攻击者将很难猜到密钥值。
在app.py文件中添加如下代码：
#导入相应模块
from flask import Flask,flash
#导入相应模块
from flask import url_for,render_template
#导入CSRFProtect模块
from flask_wtf.csrf import CSRFProtect
#导入定义的BaseLogin
#导入BaseLogin模块
from forms import BaseLogin
#导入配置文件
import config
#Flask初始化
app= Flask(__name__)
#配置文件初始化
app.config.from_object(config)
#CSRFProtect模块初始化
CSRFProtect(app)
也可以直接在config.py文件中启动CSRF模块。
CSRF_ENABLED= True
最后，我们需要在响应的html模板的Form表单中加上如下语句：
{{form.csrf_token}}
或者：
{{form.hidden_tag()}}
其中的form是views.py中对应处理函数传递过来的Form对象名称，根据具体情况会有所变化。
通过上面的配置，我们就启动了CSRF保护。
3.WTF表单登录实例¶
使用Flask-WTF进行表单验证：config.py
import os
SECRET_KEY = os.urandom(24)
CSRF_ENABLED = True
使用Flask-WTF进行表单验证：form.py
# -*- coding:utf-8 -*-
# 引入Form基类
from flask_wtf import FlaskForm
# 引入Form元素父类
from wtforms import StringField, PasswordField, validators
# 引入Form验证父类
from wtforms.validators import DataRequired, Length
# 登录表单类,继承与Form类
class BaseLogin(FlaskForm):
# 用户名
name = StringField('name', validators=[DataRequired(message="用户名不能为空"), Length(
6, 16, message='长度位于6~16之间')], render_kw={'placeholder': '输入用户名'})
# 密码
password = PasswordField('password', validators=[DataRequired(message="密码不能为空"), Length(
6, 16, message='长度位于6~16之间')], render_kw={'placeholder': '输入密码'})
使用Flask-WTF进行表单验证：app.py
from flask import Flask, flash
from flask import url_for, render_template
# from flask_wtf.csrf import CSRFProtect
# 导入定义的BaseLogin
from forms import BaseLogin
import config
app = Flask(__name__)
app.config.from_object(config)
# CSRFProtect(app)
# 定义处理函数和路由规则，接收GET和POST请求
@app.route('/login', methods=('POST', 'GET'))
def baselogin():
form = BaseLogin()
# 判断是否是验证提交
if form.validate_on_submit():
# 跳转
flash(form.name.data+'|'+form.password.data)
return '表单数据提交成功！'
else:
# 渲染
return render_template('login.html', form=form)
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run(debug=True)
使用Flask-WTF进行表单验证：login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Flask_WTF</title>
<style type="text/css">
.div1 {
height:180px;
width:380px;
border:1px solid #8A8989;
margin:0 auto;
}
.input{
display: block;
width: 350px;
height: 40px;
margin: 10px auto;
}
.button
{
background: #2066C5;
color: white;
font-size: 18px;
font-weight: bold;
height: 50px;
border-radius: 4px;
}
</style>
</head>
<body>
<div class="div1"><form action="login" method = "post">
<!--启动CSRF-->
{{form.hidden_tag()}}
{{form.name(size=16,id='name',class='input' )}}
{%for e in form.name.errors%}
<span style="color: red">{{e}}</span>
{%endfor%}
{{form.password(size=16,id='password',class='input')}}
{%for e in form.password.errors%}
<span style="color: red">{{e}}</span>
{%endfor%}
<input type="submit" value="登录" class="input button">
</form></div>
</body>
</html>
8.2 使用Flask上传文件¶
1.使用Flask上传文件的简单实现¶
Flask文件上传比较简单，需要注意以下3点要求：
一个
<form>标签被标记有enctype=multipart/form-data，并且在里面包含一个<input type=file>标签。服务端应用通过请求对象上的files字典访问文件。
使用文件的save()方法将文件永久地保存在文件系统上的某处。
注意：表单中必须要有enctype=“multipart/form-data”，否则上传文件无效。
一般可以写成
给出一个Flask框架下实现文件上传的范例：
Flask文件上传：app.py
from flask import Flask, render_template, request
import os
from werkzeug.utils import secure_filename
app = Flask(__name__)
@app.route('/', methods=['POST', 'GET'])
def hello_world():
if request.method == 'GET':
return render_template('upload.html')
else:
f = request.files['file']
basepath = os.path.join(os.path.dirname(__file__)) # 当前文件所在路径
upload_path = os.path.join(
basepath, 'static', 'uploads', secure_filename(f.filename))
# upload_path = os.path.join(basepath,,secure_filename(f.filename))
f.save(upload_path)
return "上传文件成功！！"
if __name__ == '__main__':
app.run(debug=True)
Flask文件上传：upload.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>文件上传</title>
<style type="text/css">
.div1{
height:180px;
width:380px;
border:1px solid #8A8989;
margin:0 auto;
}
.a-upload {
padding: 4px 10px;
height: 20px;
line-height: 20px;
position: relative;
cursor: pointer;
color: #888;
background: #fafafa;
border: 1px solid #ddd;
border-radius: 4px;
overflow: hidden;
display: inline-block;
*display: inline;
*zoom: 1
}
.input{
display: block;
width: 250px;
height: 30px;
margin: 10px auto;
}
.button
{
background: #2066C5;
color: white;
font-size: 18px;
font-weight: bold;
height: 30px;
border-radius: 4px;
}
</style>
</head>
<body>
<div class="div1">
<form action="" method = "post" enctype='multipart/form-data'>
<input type="file" name="file" class="input">
<input type="submit" value="上传" class="input button">
</form>
</div>
</body>
</html>
运行上面的程序之前，请先确保根目录下的static目录下已经创建好uploads目录，然后就可以运行本工程代码了。
本工程可以实现任意文件的上传，但是以中文命名的文件上传时，会出现中文名的丢失。
2.改进上传功能¶
之前实现了基本的文件上传功能，但是存在一些问题：
文件没有进行重新命名，多个用户可能可能会存在上传同名文件的问题；
没有实现文件目录的自动创建，如果网站部署时忘记创建文件保存目录，则会出现文件保存失败问题，影响用户体验；
文件上传时，没有进行必要的文件格式检验，用户可以直接将可执行文件上传到服务器上，影响到服务器的数据安全性。
Flask文件上传功能改进1：app.py
from flask import Flask, render_template, request, send_from_directory
import time
import os
from os import path
from werkzeug.utils import secure_filename
import platform
from form import UploadForm
from werkzeug.datastructures import CombinedMultiDict
app = Flask(__name__)
if platform.system() == "Windows":
slash = '\\'
else:
platform.system() == "Linux"
slash = '/'
# UPLOAD_PATH = os.path.curdir + os.path.sep + 'uploads' + os.path.sep
UPLOAD_PATH = os.path.curdir + slash + 'uploads' + slash
@app.route('/', methods=['POST', 'GET'])
def hello_world():
if request.method == 'GET':
return render_template('upload.html')
else:
if not os.path.exists(UPLOAD_PATH):
os.makedirs(UPLOAD_PATH) # 没有目录则创建目录
form = UploadForm(CombinedMultiDict([request.form, request.files]))
if form.validate():
f = request.files['file']
filename = secure_filename(f.filename) # 取得文件名字
ext = filename.rsplit('.', 1)[1] # 获取文件后缀
unix_time = int(time.time())
new_filename = str(unix_time) + '.' + ext # 对文件进行重新命名
file_url = UPLOAD_PATH+new_filename
print(new_filename)
print(UPLOAD_PATH)
print(file_url)
f.save(path.join(UPLOAD_PATH, new_filename))
return "上传文件成功！！"
else:
return "只支持jpg、png以及gif格式的文件！"
# 访问上传的文件
# 浏览器访问：http://127.0.0.1:5000/images/xxx.jpg/ 就可以查看文件了
@app.route('/images/<filename>/', methods=['GET', 'POST'])
def get_image(filename):
# 得到绝对路径，比如J:\python project\例5-3-2\uploads
dirpath = os.path.join(app.root_path, 'uploads')
print(dirpath)
print(filename)
# return send_from_directory(dirpath,filename,as_attachment=True)#为下载方式
return send_from_directory(dirpath, filename) # 为在线浏览方式
if __name__ == '__main__':
app.run(debug=True)
8.4 Session的使用¶
Session是基于Cookie实现的，保存在服务端的键值对（session[‘name’]=‘value’）中。
同时，在浏览器的Cookie中也对应一个相同的随机字符串，用来再次请求的时验证。
注意：Session是储存在服务器中的，Cookies是储存在浏览器本地中，而Flask的Session是基于Cookies，Session是经过加密保存在Cookies中。
1.Session的基本配置¶
因为Flask的Session是通过加密之后放到了Cookie中。有加密就有密钥用于解密，所以，用到了Flask的Session模块就一定要配置SECRET_KEY这个全局宏。一般将SECRET_KEY设置为24位的字符。我们可以自己设定一个随机字符串，例如：
app.config['SECRET_KEY']= 'XXXXX'
我们也可以引入OS模块中，自动产生一个24位的随机字符串函数。
这种方法有个不足之处，就是服务器每次启动之后这个SECRET_KEY的值是不一样的，会造成Session验证失效，用户只有重新登录。
Session的基本操作：app.py
from flask import Flask, session
from datetime import timedelta
import os
app = Flask(__name__)
# 设置session
app.config['SECRET_KEY'] = os.urandom(24)
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7) # 配置7天有效
@app.route('/')
def set_session():
session['username'] = 'zhangsan'
session.permanent = True
return 'Session设置成功!'
# 获取session
@app.route('/get_session')
def get_session():
username = session.get('username')
return username or 'Session为空！'
# 清除sessin
@app.route('/del_session')
def del_session():
session.pop('username')
# session.clear #清空Session
return 'Session被删除！'
if __name__ == '__main__':
app.run()
通过浏览器访问进行验证
设置session
获取session
http://127.0.0.1:5000/get_session
删除session
http://127.0.0.1:5000/del_session
再次验证session
8.5 钩子函数的使用¶
根据需要，有时候需要在正常执行代码的前、中、后时期，强行执行一段我们想要执行的功能代码，实现这种功能的函数，就称为钩子函数。钩子函数的实质就是用特定装饰器装饰的函数。下面对常用的几种钩子函数进行介绍。
请求钩子通过装饰器实现。Flask 支持以下 4 种钩子。
before_first_request：注册一个函数，在处理第一个请求之前运行
before_request：注册一个函数，在处理每一个请求的时候运行
after_request：注册一个函数，如果没有未处理的异常抛出，在每次请求之后运行
teardown_request：注册一个函数，即使有未处理的异常抛出，也在每次请求之后运行
在请求钩子函数和视图函数之间共享数据一般使用上下文全局变量g。例如，before_request 处理程序可以从数据库中加载已登录用户，并将其保存到g.user 中。随后调用视图函数时，视图函数再使用g.user 获取用户。
钩子函数的基本使用：app.py
from flask import Flask
import time
app = Flask(__name__)
# before_first_request函数一般用来作一次初始化工作
@app.before_first_request
def before_first_request():
print("这是before_first_request钩子函数")
@app.before_request
def before_request():
print("这是before_request钩子函数")
@app.after_request
def after_request(response):
print("这是after_request钩子函数")
response.headers["Content-Type"] = "application/json"
return response
@app.teardown_request
def teardown_request(e):
print("这是teardown_request钩子函数")
@app.route('/')
def hello_world():
print("您访问了首页！")
time.sleep(5)
return 'Hello World!'
if __name__ == '__main__':
app.run()
上面的代码定义并使用了4个钩子函数，其中，
before_first_request()钩子函数会在第一次请求之前被执行，
before_reques()钩子函数对每次请求都会执行，
after_request()和teardown_request()会在每次请求之后执行。
在地址栏中输入如下地址：http://127.0.0.1:5000/然后回车，运行结果如下：
Press CTRL+C to quit
这是before_first_request钩子函数
这是before_request钩子函数
您访问了首页！
这是after_request钩子函数
这是teardown_request钩子函数
127.0.0.1 - - [07/Sep/2022 16:54:42] "GET / HTTP/1.1" 200 -
这是before_request钩子函数
您访问了首页！
这是after_request钩子函数
这是teardown_request钩子函数
9.访问数据库¶
9.1 MySQL数据库安装¶
软件自行安装,可参考
https://www.runoob.com/mysql/mysql-install.html
docker部署MySQL 5.7 数据库
$ docker run --name mysql -d -e MYSQL_RANDOM_ROOT_PASSWORD=yes \
-e MYSQL_DATABASE=flasky -e MYSQL_USER=flasky \
-e MYSQL_PASSWORD=<database-password> \
mysql/mysql-server:5.7
Docker Compose部署Mysql 5.7
version: "3"
services:
mysql:
image: mysql:5.7
restart: always
hostname: mysql-master
container_name: "mysql57-giteeci"
ports:
- 33006:3306
environment:
TZ: Asia/Shanghai
MYSQL_ROOT_PASSWORD: oschina123
MYSQL_DATABASE: gitlabhq_production
MYSQL_USER: giteeci
MYSQL_PASSWORD: 123456
MYSQL_HOST: "%"
volumes:
- /data/mysql57:/var/lib/mysql
- ./mysql.cnf:/etc/mysql/conf.d/mysql.cnf
command:
--server_id=100
--log-bin=mysql-bin
--sync_binlog=1
--binlog-ignore-db=mysql
--binlog_format=mixed
--expire_logs_days=7
--default-authentication-plugin=mysql_native_password
--character-set-server=utf8mb4
--collation-server=utf8mb4_general_ci
--explicit_defaults_for_timestamp=true
--lower_case_table_names=1
--sql_mode=NO_ENGINE_SUBSTITUTION,STRICT_TRANS_TABLES
#--sql_mode=STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION
9.2 Python数据库框架MySQL-Python¶
注意：Python 2.x用MySQL-Python，从Python 3.x起，不再支持MySQL-Python。
在Python 3.x中要使用MySQLdb,可以使用PyMySQL，在Python 3.x中PyMySQL替代了MySQLdb。
安装PyMySQL可以使用下面命令：
（venv）$ pip install pyMySQL
此外，在数据库初始化代码中，需要配置编写下面代码：
import pyMySQL
#手动指定将MySQLdb转给PyMySQL处理
pyMySQL.install_as_MySQLdb()
1.通过Python操作数据库对象¶
利用PyMySQL框架访问数据库：app.py
from flask import Flask
import pymysql
pymysql.install_as_MySQLdb() # 手动指定将MySQLdb转给pymysql处理
app = Flask(__name__)
conn = pymysql.Connect( # 配置数据库连接
host='127.0.0.1',
port=3306,
user='root',
passwd='root',
db='demo_01',
charset='utf8'
)
cursor = conn.cursor()
sql = "select * from user"
cursor.execute(sql)
results = cursor.fetchall()
for row in results:
username = row[0]
email = row[1]
# 打印结果
print(username)
cursor.close() # 关闭游标
conn.close() # 关闭连接
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
2.通过PyMySQL进行更新数据操作¶
增加数据
利用PyMySQL框架向数据库中添加数据：app.py
# -*- coding: UTF-8 -*-
from flask import Flask
import pymysql # 引入mysql库：
pymysql.install_as_MySQLdb()
app = Flask(__name__)
conn = pymysql.Connect( # 建立数据库连接
host='127.0.0.1',
port=3306,
user='root',
passwd='root',
db='demo_01',
charset='utf8'
)
cursor = conn.cursor() # 使用cursor()方法获取操作游标
# 使用try---except进行异常处理
try:
sql_insert = "insert into user(username,email) values ('yanghong1','22222@qq.com')"
sql_insert1 = "insert into user(username,email) values ('caoxiao1','33333@qq.com')"
cursor.execute(sql_insert)
cursor.execute(sql_insert1)
conn.commit()
except Exception as e:
# print e
print(e)
conn.rollback()
cursor.close()
conn.close() # 关闭数据库连接
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
conn.commit()语句放在两条插入语句执行之后，并放在try语句块内，如果产生异常，可以使用：
conn.rollback()
该语句可以使事务回滚，退回到没有执行数据插入之前。
cursor.execute(“sql语句”)表示使用execute方法执行SQL语句。
所用的SQL操作都放在cursor.execute()方法中来执行。
修改数据
修改数据库中某条记录，同样是先建立数据库连接，然后获取游标对象，通过cursor.execute()来执行SQL语句，使用conn.commit()方法来提交事务，让修改数据操作真正生效。
删除数据
要删除数据库中的某条记录，同样是先建立数据库连接，然后获取游标对象，通过cursor.execute()来执行SQL语句，使用conn.commit()方法来提交事务，让删除数据操作真正生效。
9.3 初识Flask-SQLAlchemy¶
SQLAlchemy是一个基于Python实现的ORM（Object Relational Mapping，对象关系映射）框架。该框架建立在DB API之上，使用关系对象映射进行数据库操作。
简言之便是将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
它的核心思想于在于将关系数据库表中的记录映射成为对象，以对象的形式展现，程序员可以把对数据库的操作转化为对对象的操作。
注意：Flask-SQLAlchemy框架的实质在于以操作对象的方式对数据库操作。
1.SQLAlchemy的安装¶
（venv）$ pip install flask-sqlalchemy
注意：Python 2版本中需要安装MySQLdb，可以使用pip install flask-MySQLdb命令来完成安装。 Python 3中不再支持MySQLdb，推荐使用PyMySQL。
2.为什么使用ORM¶
当需要实现一个应用程序时（不使用O/R Mapping），我们可能会写特别多的数据访问层的代码，从数据库保存、删除、读取对象信息，而这些代码都是重复的。
而使用ORM则会大大减少重复性代码。
对象关系映射（Object Relational Mapping，ORM），主要实现程序对象到关系数据库数据的映射。
简单：ORM以最基本的形式建模数据。比如ORM会将MySQL的一张表映射成一个类（模型），表的字段就是这个类的成员变量。
精确：ORM使所有的MySQL数据表都按照统一的标准精确地映射成一个类，使系统在代码层面保持准确统一。
易懂：ORM使数据库结构文档化，程序员可以把大部分精力用在Web功能的开发和实现上，而不需要花费时间和精力在底层数据库驱动上。
易用：ORM包含对持久类对象进行CRUD操作的API，例如create()、update()、save()、load()、find()、find_all()和where()等，也就是将SQL查询全部封装成了编程语言中的函数，通过函数的链式组合生成最终的SQL语句。通过这种封装避免了不规范、冗余、风格不统一的SQL语句，可以避免很多人为Bug，方便编码风格的统一和后期维护。
综上所述，使用ORM框架的最大优点是解决了重复读取数据库的问题，使程序员高效开发成为可能。
最大的不足之处在于会牺牲程序的执行效率，特别是处理多表联查、where条件复杂之类的查询时，ORM的语法会变得复杂。
3.Flask-SQLAlchemy初始化¶
使用Flask-SQLAlchemy框架初始化实例：config.py
USERNAME = 'root'
PASSWORD = 'root'
HOST = '127.0.0.1'
PORT = '3306'
DATABASE = 'db_demo1'
DB_URI = 'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(
USERNAME, PASSWORD, HOST, PORT, DATABASE)
SQLALCHEMY_DATABASE_URI = DB_URI
# 动态追踪修改设置，如未设置只会提示警告
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO = True
注意：SQLALCHEMY_TRACK_MODIFICATIONS=False必须设置，可以为Flase或者True，表示动态追踪修改设置。如果没有设置，会给出警告信息。
使用Flask-SQLAlchemy框架初始化实例：db_demo1.py
# encoding:utf8
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import config # 导入配置文件
app = Flask(__name__)
app.config.from_object(config) # 配置文件实例化
# 初始化一个对象
db = SQLAlchemy(app)
# 测试数据库链接是否成功
db.create_all()
@app.route('/')
def index():
return "index"
if __name__ == '__main__':
app.run(debug=True)
运行上述代码，如果没有报错，则说明配置成功。可以将DATABASE=’db_demo1’的内容修改为一个不存在的数据库，则可以看到有报错信息输出。
4.模型与表映射方法1¶
使用Flask-SQLAlchemy框架创建数据库表：app.py
# encoding:utf8
from flask import Flask # 导入Flask模块
from flask_sqlalchemy import SQLAlchemy # 导入SQLAlchemy模块
import config # 导入配置文件
from datetime import datetime # 导入datetime模块
app = Flask(__name__) # Falsk初始化
app.config.from_object(config) # 配置文件实例化
# 初始化一个数据库连接对象
db = SQLAlchemy(app)
# 测试数据库链接是否成功
# 建立表
class Book(db.Model):
__tablename__ = 'book' # 表的别名
id = db.Column(db.Integer, primary_key=True, autoincrement=True) # id号
title = db.Column(db.String(50), nullable=False) # 书名
publishing_office = db.Column(db.String(100), nullable=False) # 出版社
isbn = db.Column(db.String(100), nullable=False) # isbn号
storage_time = db.Column(db.DateTime, default=datetime.now) # 入库时间
db.create_all() # 对象映射
@app.route('/')
def index():
return "index"
if __name__ == '__main__':
app.run(debug=True)
使用Flask-SQLAlchemy框架创建数据库表：config.py
# encoding:utf8
USERNAME = 'root'
PASSWORD = 'root'
HOST = '127.0.0.1'
PORT = '3306'
DATABASE = 'db_demo1'
DB_URI = 'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(
USERNAME, PASSWORD, HOST, PORT, DATABASE)
SQLALCHEMY_DATABASE_URI = DB_URI
# 动态追踪修改设置，如未设置只会提示警告
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO = True
Python语言的3.x完全不向前兼容，导致在Python 2.x中可以正常使用的库到了Python 3中就不支持MySQLdb数据库驱动了。
给出的解决方法是，使用PyMySQL作为Python 3环境下MySQLdb的替代数据库驱动，可以进入命令行，使用pip安装PyMySQL。
安装完毕后，如果你继续要使用MySQLdb，那么请加上一句pyMySQL.install_as_MySQLdb()即可。
注意：添加pyMySQL.install_as_MySQLdb();一般是主app对应的文件中。
最常用的SQLAlchemy列类型
类型名 |
Python类型 |
说明 |
|---|---|---|
Integer |
int |
普通整数，通常是 32 位 |
SmallInteger |
int |
取值范围小的整数，通常是 16 位 |
BigInteger |
int 或 long |
不限制精度的整数 |
Float |
float |
浮点数 |
Numeric |
decimal.Decimal |
定点数 |
String |
str |
变长字符串 |
Text |
str |
长文本 text |
Unicode |
unicode |
变长 Unicode 字符串 |
UnicodeText |
unicode |
变长 Unicode 字符串，对 较长或不限长度的字符串做了优化 |
Boolean |
bool |
布尔值 |
Date |
datetime.date |
日期 |
Time |
datetime.time |
时间 |
DateTime |
datetime.datetime |
日期和时间 |
Interval |
datetime.timedelta |
时间间隔 |
Enum |
str |
一组字符串 |
PickleType |
任何 Python 对象 |
自动使用 Pickle 序列化 |
LargeBinary |
str |
二进制 blob |
最常用的SQLAlchemy列选项
选项名 |
说明 |
|---|---|
primary_key |
如果设为 True，列为表的主键 |
unique |
如果设为 True，列不允许出现重复的值 |
index |
如果设为 True，为列创建索引，提升查询效率 |
nullable |
如果设为 True，列允许使用空值；如果设为 False，列不允许使用空值 |
autoincrement |
如果设为 True，主键设置自增长 |
default |
为列定义默认值 |
模型定义示例
class Members(db.Model):
__tablename__ = 'jq_member'
uid = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(50), nullable=False, unique=True) # 用户名不能为空,而且必须是唯一的
_password = db.Column(db.String(100), nullable=False) # 密码不能为空
email = db.Column(db.String(50), nullable=False, unique=True) # 用户邮箱不能为空，而且必须是唯一的
vatar=db.Column(db.String(80),nullable=True)#用户头像
nickname=db.Column(db.String(50),nullable=True)#用户昵称
sex = db.Column(db.String(2), default=0) # 性别
telephone = db.Column(db.String(11)) # 电话
status = db.Column(db.Integer) # 状态
def __init__(self,username,password,email):
self.username=username
self.password=password
self.email=email
#获取密码
@property
def password(self):
return self._password
#设置密码
@password.setter
def password(self,raw_password):
self._password=generate_password_hash(raw_password)#密码加密
#检查密码
def check_password(self,raw_password):
result=check_password_hash(self.password,raw_password)#
return result
新增一条记录，可以使用下面的方法。在执行下面的方法之前先把db.create_all()这行注释掉。注释以后变为：
# db.create_all() # 对象映射
book1 = Book(id='001', title='人工智慧', publishing_office='高等教育', isbn='123333')
# 通过db.session.add(book1)把对象添加到会话中
db.session.add(book1)
# 通过db.session.commit()完成事务的提交。
db.session.commit()
使用Flask-SQLAlchemy框架添加数据实例：app.py
from flask import Flask # 导入Flask模块
from datetime import datetime # 导入datetime模块
import config # 导入配置文件
from flask_sqlalchemy import SQLAlchemy # 导入SQLAlchemy模块
import pymysql # 导入pymysql
pymysql.install_as_MySQLdb() # pymysql替代MySQLdb
app = Flask(__name__)
app.config.from_object(config)
db = SQLAlchemy(app) # 读取配置参数，将和数据库相关的配置加载到SQLAlchemy对象中
db.init_app(app) # 初始化数据库连接文件
# 建立表
class Book(db.Model):
__tablename__ = 'book' # 表的别名
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
title = db.Column(db.String(50), nullable=False) # 书名
publishing_office = db.Column(db.String(100), nullable=False) # 出版社
isbn = db.Column(db.String(100), nullable=False) # isbn号
storage_time = db.Column(db.DateTime, default=datetime.now) # 入库时间
db.create_all() # 对象映射
# 以上代码同样创建了一个表，下面试着增加一条记录，再增加下面代码：
book1 = Book(id='005', title='人工智能导论',
publishing_office='高等教育出版社', isbn='9787040479843')
db.session.add(book1) # 把对象添加到会话中
db.session.commit() # 提交事务
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
运行app.py文件，会在数据库中创建一条记录。
5. 模型与表映射方法2¶
这里底层驱动使用了PyMySQL，使用Flask-SQLAlchemy创建数据库，并成功地向数据库中增加了数据。
使用Flask-SQLAlchemy框架添加数据实例：app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import config
app = Flask(__name__)
app.config.from_object(config)
db = SQLAlchemy(app)
class Book(db.Model):
__tablename__ = 'book'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
title = db.Column(db.String(50), nullable=False)
publishing_office = db.Column(db.String(100), nullable=False)
price = db.Column(db.String(30), nullable=False)
isbn = db.Column(db.String(50), nullable=False)
storage_time = db.Column(db.DateTime, default=datetime.now) # 入库时间
# db.create_all()#如果没有创建表，请先执行db.create_all（）方法创建表
# 添加数据的路由
@app.route('/add')
def add():
book1 = Book(title='Python基础教程1（第3版）', publishing_office='人民邮电出版社',
price='68.30', isbn='9787115474889')
# book2 = Book(title='Python游戏编程快速上手 第4版', publishing_office='人民邮电出版社', price='54.50', isbn='9787115466419')
# book3 = Book(title='JSP+Servlet+Tomcat应用开发从零开始学', publishing_office='清华大学出版社', price='68.30',isbn='9787302384496')
db.session.add(book1)
# db.session.add(book2)
# db.session.add(book3)
db.session.commit()
return '添加数据成功！'
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run(debug=True)
保存并执行app.py文件，然后进入数据库查看数据是否插入成功。
6.创建表 简单的增删改查¶
创建表
@app.route('/create_table/')
def create_table():
db.create_all()
return 'create_table'
删除表
@app.route('/drop_table/')
def drop_table():
db.drop_all()
return 'drop_table'
添加一条数据 add
@app.route('/insert_one/')
def insert_one():
try:
u = User(username='张三',password_hash='zhangsan123456',age=20,sex=False,email='948807313@qq.com',confirm=True)
db.session.add(u)
db.session.commit() #提交 默认开启了事物
except:
db.session.rollback() #回滚
return 'insert_one'
添加多条数据 add_all
#添加多条数据
@app.route('/insert_many/')
def insert_many():
try:
u1 = User(username='李四')
u2 = User(username='王五')
db.session.add_all([u1,u2])
db.session.commit() #提交 默认开启了事物
except:
db.session.rollback() #回滚
return 'insert_many'
查询一条数据 get
@app.route('/select_one/')
def select_one():
u = User.query.get(1)
print(u.username)
print(u.age)
print(u.sex)
return 'select_one'
修改一条数据 add
#修改数据
@app.route('/update/')
def update():
u = User.query.get(1)
u.username = '赵六'
u.age = 30
u.sex = True
db.session.add(u)
db.session.commit()
return 'update'
删除一条数据 delete
#删除数据
@app.route('/delete/')
def delete():
u = User.query.get(1)
db.session.delete(u)
db.session.commit()
return 'delete'
更新内容参考
7.数据的增、删、改、查¶
数据的增、删、改、查等操作，必须确保数据库已经实现建立好，然后使用db.session.add(对象名称)方法，就可以实现数据的插入操作。
使用Flask-SQLAlchemy框架添加多条数据实例：app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import config
app = Flask(__name__)
app.config.from_object(config)
db = SQLAlchemy(app)
class Book(db.Model):
__tablename__ = 'book'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
title = db.Column(db.String(50), nullable=False)
publishing_office = db.Column(db.String(100), nullable=False)
price = db.Column(db.String(30), nullable=False)
isbn = db.Column(db.String(50), nullable=False)
storage_time = db.Column(db.DateTime, default=datetime.now) # 入库时间
db.create_all()
@app.route('/add')
def add():
book1 = Book(title='Python基础教程（第3版）', publishing_office='人民邮电出版社',
price='68.30', isbn='9787115474889')
book2 = Book(title='Python游戏编程快速上手 第4版', publishing_office='人民邮电出版社',
price='54.50', isbn='9787115466419')
book3 = Book(title='JSP+Servlet+Tomcat应用开发从零开始学',
publishing_office='清华大学出版社', price='68.30', isbn='9787302384496')
db.session.add(book1)
db.session.add(book2)
db.session.add(book3)
db.session.commit()
return '添加数据成功！'
@app.route('/select')
def select():
# result=Book.query.filter(Book.id=='1').first()
result = Book.query.filter(Book.publishing_office == '人民邮电出版社').all()
for books in result:
print(books.title, books.publishing_office)
return "查询数据成功！"
@app.route('/edit')
def edit():
book1 = Book.query.filter(Book.id == '2').first() # 查询出id=2的记录
book1.price = 168 # 修改价格
db.session.commit()
return "修改数据成功！"
@app.route('/delete')
def delete():
book1 = Book.query.filter(Book.id == '9').first()
db.session.delete(book1)
db.session.commit()
return '删除数据成功'
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run(debug=True)
创建cofing.py文件，该文件为数据库连接配置文件
#encoding:utf-8
HOST = '127.0.0.1'
PORT = '3306'
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'books_01'
DB_URI = 'mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset-utf8'.format(username=USERNAME,
password=PASSWORD,
host=HOST,
port=PORT,
db=DATABASE)
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
9.4 创建一对一的关系表¶
数据库实体间有3种关联关系：一对一、一对多，以及多对多。
一个学生只有一个身份证号码，构成了一对一关系；
一个班级有多名学生，构成了一对多的关系；
课程和学生的关系就构成了多对多关系，一个学生可以选修多门课程，一门课程对应多个学生。
比如人和身份证的一对一关系，图书管理系统中的学生和借书证的一对一关系、学生和具体借到的某一本书也是一对一关系等。
一对一关系主要在relationship方法中，使用uselist=False来约束，如果查询得到的是一个列表，那么就使用uselist=False禁用列表，这样最终查询到的结果就是唯一的，就构成了一对一关系。
注意：uselist=False表示不使用列表，使用标量值。
下面主要图书管理系统中的用户和借书证构成了一对一关系为例进行讲解。
主要文件为：
config.py
app.py
使用Flask-SQLAlchemy创建一对一关系实例：app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import config
app = Flask(__name__)
app.config.from_object(config)
db = SQLAlchemy(app)
# 定义用户表
class User(db.Model):
__tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(50), nullable=False) # 用户名
password = db.Column(db.String(50), nullable=False) # 密码
phone = db.Column(db.String(11), nullable=False) # 电话
email = db.Column(db.String(30), nullable=False) # 邮箱
reg_time = db.Column(db.DateTime, default=datetime.now) # 注册时间
# 定义借书证表
class Lib_card(db.Model):
__tablename__ = 'lib_card'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
card_id = db.Column(db.Integer, nullable=False) # 借书证id
papers_type = db.Column(db.String(50), nullable=False) # 何种证件办理
borrow_reg_time = db.Column(db.DateTime, default=datetime.now) # 证件办理时间
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
users = db.relationship('User', backref=db.backref('cards'), uselist=False)
db.create_all()
@app.route('/add')
def add():
# 添加两条用户数据
user1 = User(username="张三", password="111111",
phone="13888888888", email="10086@qq.com")
user2 = User(username="李四", password="123456",
phone="13777777777", email="10000@qq.com")
db.session.add(user1)
db.session.add(user2)
# 添加两条借书证信息
card1 = Lib_card(card_id='18001', user_id='1', papers_type='身份证')
card2 = Lib_card(card_id='18002', user_id='2', papers_type='身份证')
db.session.add(card1)
db.session.add(card2)
db.session.commit()
return '添加数据成功！'
@app.route('/select')
def select():
user = User.query.filter(User.username == '张三').first()
art = user.cards
for k in art:
print(k)
print(k.card_id)
card = Lib_card.query.filter(Lib_card.card_id == '18001').first()
user = card.users
print(user.username)
return "查询数据成功！"
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run(debug=True)
config.py
# encoding:utf-8
HOST = '127.0.0.1'
PORT = '3306'
USERNAME = 'root'
PASSWORD = '123456'
DATABASE = 'library'
DB_URI = 'mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset-utf8'.format(username=USERNAME,
password=PASSWORD,
host=HOST,
port=PORT,
db=DATABASE)
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
注意：一个表（模型）的定义必须要定义一个主键，这个主键一般为id。
在定义了Lib_card类后，申明了一个外键，并且在relationship方法中使用uselist=False来约束其关系。
user_id=db.Column(db.Integer,db.ForeignKey(‘user.id’))表示创建一个外键，类型要跟主表一样，通过db.ForeignKey(“user.id”)与主表绑定users=db.relationship(‘User’，backref=db.backref(‘cards’)；
uselist=False表示user可以根据Lib_card中的借书证查找到用户表中的信息
backref=“cards”表示用户表可以直接通过cards查找到该用户下的借书证号码。
9.5 创建一对多的关系表¶
生活中典型的一对多关系有哪些？一个年级对应多个平行班级，一个网购用户对应多个订单，一个家庭对应多个家庭成员等，这些是典型的一对多关系。
下面以一个典型的一对多关系为例，即一个作者可以编著多本图书的一对多关系为例，来揭晓用flask-sqlalchemy是如何实现一对多关系的。
使用Flask-SQLAlchemy创建一对多关系实例：app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import config
app = Flask(__name__)
app.config.from_object(config)
db = SQLAlchemy(app)
# 定义用户表
# 定义模型类-作者类
class Writer(db.Model):
__tablename__ = 'writer'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), nullable=False)
# 设置relationship属性方法，建立模型关系，第一个参数为多方模型的类名，添加backref可以实现多对一的反向查询
books = db.relationship('Book', backref='writers')
# 定义模型类-图书类
class Book(db.Model):
__tablename__ = 'books'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(50), nullable=False)
publishing_office = db.Column(db.String(100), nullable=False)
isbn = db.Column(db.String(50), nullable=False)
# 设置外键指向一方的主键，建立关联关系
writer_id = db.Column(db.Integer, db.ForeignKey('writer.id'))
db.create_all()
@app.route('/add')
def add():
# 添加两条作者数据
user1 = Writer(name="李兴华")
user2 = Writer(name="Sweigart")
db.session.add(user1)
db.session.add(user2)
# 添加两条图书信息
book1 = Book(title='名师讲坛——Java开发实战经典（第2版）',
publishing_office='清华大学出版社', isbn='9787302483663', writer_id='1')
book2 = Book(title='android开发实战', publishing_office='清华大学出版社',
isbn='9787302281559', writer_id='1')
book3 = Book(title='Python游戏编程快速上手', publishing_office='人民邮电大学出版社',
isbn='9787115466419', writer_id='2')
db.session.add(book1)
db.session.add(book2)
db.session.add(book3)
db.session.commit()
return '添加数据成功！'
@app.route('/select')
def select():
writer = Writer.query.filter(Writer.id == '1').first()
book = writer.books
for k in book:
print(k)
print(k.title)
book = Book.query.filter(Book.id == '1').first()
writer = book.writers
print(writer.name)
return "查询数据成功！"
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
主要是通过设置外键指向一方的主键，让作者和图书建立关联关系。
通过创建一个外键，类型要与想要关联的表的数据类型一样，这里通过db.ForeignKey(‘writer.id’)来实现。
9.6 创建多对多的关系表¶
生活中典型的多对多关系比较多，一个班级对应多个老师，一个老师对应多个班级。
又比如，一个学生对应多门课程，一门课程对应多个学生等，这些是典型的多对多关系。
下面以一个典型的多对多关系为例讲解，即图书和图书标签（上架建议标签）的数据库。
很显然，图书可能对应着多个标签，不能再使用外键来描述其关系了。
同样，也不能在标签表中加入一个指向图书的外键，因为一个标签可以对应着多本图书，两侧都需要一组外键。
处理多对多表问题时，解决方法是添加第3张表，这个表称为关联表或中间表。
数据库中的多对多关联关系一般需采用中间表的方式处理，将多对多转化为两个一对多。
注意：关联表或中间表不要定义成类，不要使用class，在使用db.Table()方法定义时，应该使用两个以上的外键。
使用Flask-SQLAlchemy创建多对多关系实例：app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import config
app = Flask(__name__)
app.config.from_object(config)
db = SQLAlchemy(app)
book_tag = db.Table('book_tag',
db.Column('book_id', db.Integer, db.ForeignKey(
'book.id'), primary_key=True),
db.Column('tag_id', db.Integer, db.ForeignKey(
'shelfing.id'), primary_key=True)
)
# 定义模型类-图书类
class Book(db.Model):
__tablename__ = 'book'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
name = db.Column(db.String(50), nullable=False)
# 设置relationship属性方法，建立模型关系，第一个参数为多方模型的类名，secondary代表中间表
# 第三个参数表示反向引用
tags = db.relationship('Shelfing', secondary=book_tag,
backref=db.backref('books'))
# 定义模型类-图书上架建议（标签）类
class Shelfing(db.Model):
__tablename__ = 'shelfing'
id = db.Column(db.Integer, primary_key=True, nullable=False)
tag = db.Column(db.String(50), nullable=False)
db.create_all()
@app.route('/add')
def add():
book1 = Book(name='Java开发')
book2 = Book(name='Python游戏编程快速上手')
book3 = Book(name='文艺范')
tag1 = Shelfing(tag='文艺')
tag2 = Shelfing(tag='计算机')
tag3 = Shelfing(tag='技术')
book1.tags.append(tag2)
book1.tags.append(tag3)
book2.tags.append(tag3)
book3.tags.append(tag1)
db.session.add_all([book1, book2, book3, tag1, tag2, tag3])
db.session.commit()
return '数据添加成功！'
@app.route('/select')
def select():
# 查询Java开发这本书的上架标签
book = Book.query.filter(Book.name == 'Java开发').first()
tag = book.tags
for k in tag:
print(k.tag)
# 查询标签=技术的所有图书
tag = Shelfing.query.filter(Shelfing.tag == '技术').first()
book = tag.books
for k in book:
print(k.name)
return '查询成功！'
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
使用http://127.0.0.1:5000/地址访问，
在数据中新增表，使用http://127.0.0.1:5000/add新增测试数据，
使用http://127.0.0.1:5000/selcet进行测试数据的查询。
db.session.add_all()表示批量作insert操作，在插入单条数据的时候使用db.session.add()，
在插入多条数据的时候使用db.session.add_all()。
注意：使用http://127.0.0.1:5000/add新增测试数据时，刷新一次网页，会增加一次数据。
多对多关系仍使用定义一对多关系的db.relationship()方法进行定义。
在Book类中，我们使用了tags=db.relationship(‘Shelfing’,secondary=book_tag,backref=db.backref(‘books’))来定义。db.relationship()一共给出了3个参数；
第1个参数Shelfing为多方模型的类名；
第2个参数secondary表示在多对多关系中；必须把secondary参数设为关联表book_tag；
第3个参数backref表示反向引用的名称。
假如拿到了一个标签tag，怎么拿到标签下的所有图书呢？这时用books，表示反向引用。多对多关系可以在任何一个类中定义，backref参数会处理好关系的另一侧。
关联表就是一个简单的表，不是模型，不要定义为类，SQLAlchemy会自动接管这个表。
10.Flask-Script工具的使用¶
Flask-Script的作用是可以通过命令行的形式来操作Flask。例如，通过命令跑一个开发版本的服务器、设置数据库，定时任务等。
10.1 安装Flask-Script并初始化¶
（venv）$ pip install Flask-Script
首先，创建一个Python模板运行命令脚本，可起名为manager.py。
在该文件中必须有一个Manager实例，Manager类追踪所有在命令行中调用的命令和处理过程的调用与运行情况；Manager只有一个参数——Flask实例，也可以是一个函数或其他的返回Flask实例；
创建Manager实例时需要用到Flask对象。
调用manager.run()启动Manager实例接收命令行中的命令。
# 导入Manager模块
from flask_script import Manager
# Flask初始化
app= Flask(__name__)
# 这里一般是设置你的app操作
manager= Manager(app)
if __name__== "__main__":
# 创建Manager的实例manager
manager.run()
10.2 Command子类创建命令¶
定义一个子类继承自Command类，然后在manager.add_command()方法中可以将这个类定义成一些命令，此时的命令应该用单引号或双引号引起来。具体用法如下：
Flask-Scrip工具使用实例：manager.py
# -*- coding:utf-8 -*-
from flask_script import Manager, Server # 导入相应模块
from flask_script import Command # 导入Command模块
from app import app # 导入app
manager = Manager(app) # 创建Manager的实例manager
class Hello(Command): # 定义Hello类
'hello world'
def run(self): # 定义函数run
print('hello world') # 打印输出
# 自定义命令一：
manager.add_command('hello', Hello()) # 把Hello()子类定义为命令hello
# 自定义命令二：
manager.add_command("runserver", Server()) # 命令是runserver
if __name__ == '__main__':
manager.run()
Python flask flask_script 报错：ModuleNotFoundError: No module named ‘flask._compat‘解决方法
原因：
可能是更新了Flask版本问题，导致Flask._compat的目录发生变化。
解决方法：
使用如下命令降级；
$ python -m pip uninstall flask
$ python -m pip install flask==1.1.4
$ python -m pip install markupsafe==2.0.1
或者
不降级则可以尝试修改一下
flask_script/__init__.py文件中
from ._compat import text_type
改成
from flask_script._compat import text_type
注意：这里定义的命令hello和runserver使用单引号或双引号均可。
执行如下命令：
$ python manager.py
usage: manager.py [-?] {hello,runserver,shell} ...
positional arguments:
{hello,runserver,shell}
hello hello world
runserver Runs the Flask development server i.e. app.run()
shell Runs a Python shell inside Flask application context.
optional arguments:
-?, --help show this help message and exit
$ python manager.py hello
hello world
$ python manager.py runserver
* Serving Flask app 'app'
* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
* Running on http://127.0.0.1:5000
Press CTRL+C to quit
10.3 使用Command实例的@command修饰符¶
@command添加命令使用实例：manager.py
# -*- coding:utf-8 -*-
from flask_script import Manager, Server # 导入相应模块
from flask_script import Command # 导入Command模块
from app import app # 导入app
manager = Manager(app) # 创建Manager的实例manager
@manager.command # 使用装饰器
def hello(): # 定义函数hello()
'hello world'
print('hello world') # 打印输出
if __name__ == '__main__':
manager.run()
使用Command实例的@command修饰符方法来创建命令和Command类创建命令的运行方式相同。
在命令行输入以下命令，结果如下：
$ python manager.py hello
hello world
10.4 使用Command实例的@option修饰符创建命令¶
如果想要添加多个命令，建议使用@option修饰符创建命令，可以有多个@option选项参数。
@option修饰符创建命令使用实例：app.py
# -*-coding:utf8-*-
from flask_script import Manager # 导入Manager模块
from app import app # 导入app
manager = Manager(app) # 创建Manager的实例manager
# 创建命令
@manager.option('-n', '--name', dest='name', help='Your name', default='world')
@manager.option('-u', '--url', dest='url', default='www.csdn.com') # 创建命令
def hello(name, url): # 定义函数hello 'hello world or hello <setting name>'
print('hello', name) # 打印输出
print(url) # 打印输出
if __name__ == '__main__':
manager.run()
输入python manager.py hello，运行结果如下：
python manager.py hello
>hello world
>www.csdn.com
再输入python manager.py hello-n sissiy-u www.google.com，运行结果如下：
python manager.py hello-n sissiy-u www.google.com
> hello sissiy
> www.google.com
10.5 Flask循环引用¶
将数据库的model单独写在一个文件中，model类需要继承db.Model，而主文件中需要注册数据库，这样会导致在主文件中引用model，在model中引用主文件的db，造成循环引用。
基本的文件结构如下：
app.py
models.py
循环引用问题的出现实例：app.py
# encoding:utf-8
from flask import Flask # 导入Flask模块
from models import User # 导入user模块
from flask_sqlalchemy import SQLAlchemy # 导入SQLAlchemy模块
import config # 导入配置文件
app = Flask(__name__) # Flask初始化
app.config.from_object(config) # 初始化配置文件
db = SQLAlchemy(app) # 获取配置参数，将和数据库相关的配置加载到SQLAlchemy对象中
# 创建表和字段
db.create_all() # 关系映射
@app.route('/') # 定义路由
def hello_world(): # 定义hello_world函数
return 'Hello World!'
if __name__ == '__main__':
app.run()
循环引用问题的出现实例：models.py
# -*- coding:utf-8 -*-
from app import db # 导入db 对象
class User(db.Model): # 定义User类
__tablename__ = 'user' # 表的别名
id = db.Column(db.Integer, primary_key=True, autoincrement=True) # 定义id字段
username = db.Column(db.String(50), nullable=False) # 定义username字段
password = db.Column(db.String(100), nullable=False) # 定义password字段
telephone = db.Column(db.String(11), nullable=False) # 定义telephone字段
但此时会出现循环引用的错误，也就是你需要我的，我需要你的。解决办法是把db单独写在一个文件中，
然后主文件中引用db并注册，model也引用db，这样就不会造成了循环引用问题。
注意：出现循环引用的问题往往意味着代码的布局有问题，可以把需要import的资源提取到一个第三方文件中。
新建一个external.py文件结构如下：
app.py
models.py
external.py
循环引用问题的解决方法实例：external.py
# encoding:utf-8
from flask_sqlalchemy import SQLAlchemy # 导入SQLAlchemy模块
# 此时先不传入app
db = SQLAlchemy()
HOST = '127.0.0.1'
PORT = '3306'
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'library'
DB_URI = 'mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset-utf8'.format(username=USERNAME,
password=PASSWORD,
host=HOST,
port=PORT,
db=DATABASE)
# 行数据库连接的实例化
SQLALCHEMY_DATABASE_URI = DB_URI
# 关闭动态跟踪
SQLALCHEMY_TRACK_MODIFICATIONS = False
修改后的app.py文件代码如下：
循环引用问题的解决方法实例：app.py
# encoding:utf-8#设定编码
from flask import Flask # 导入Flask模块
from models import User # 导入User模块
from external import db # 导入db模块
import external # 导入extenal模块
app = Flask(__name__) # Flask初始化
app.config.from_object(external) # 配置文件初始化
db.init_app(app) # 初始化数据库连接
# 开始创建表和各个字段
with app.app_context(): # 手动创建应用上下文
db.create_all() # 建立映射
@app.route('/') # 定义路由
def hello_world(): # 定义视图函数
return 'Hello World!'
if __name__ == '__main__':
app.run(debug=True)
循环引用问题的解决方法实例：models.py
from app import db # 导入db 对象
class User(db.Model): # 定义User类
__tablename__ = 'user' # 表的别名
id = db.Column(db.Integer, primary_key=True, autoincrement=True) # 定义id字段
username = db.Column(db.String(50), nullable=False) # 定义username字段
password = db.Column(db.String(100), nullable=False) # 定义password字段
telephone = db.Column(db.String(11), nullable=False) # 定义telephone字段
11. Flask-Migrate实现数据迁移¶
我们在系统开发过程中，经常碰到需要更新数据库中的表或修改字段等操作。如果通过手工编写alter SQL脚本进行处理，经常会发现遗漏，而且修改起来不太方便。
同时，由于在Python中采用db.create_all修改字段时，不会自动将更改写入数据库的表中，只有数据表不存在时，Flask_SQLAlchemy才会创建数据库，所以必须删除数据库相关表，然后重新运行db.create_all才会重新生成表，这肯定与实际情况不符合。
现在我们可以使用Flask-Migrate迁移框架来解决这个问题。使用Flask-Migrate数据库迁移框架，可以保证数据库结构在发生变化时，改变数据库结构不至于丢失数据库的数据。
11.1 安装Flask-Migrate插件¶
(venv) $ pip install flask-migrate
11.2 使用Flask-Migrate的步骤¶
（1）修改Flask app部分的代码，以增加与Migrate相关的Command。
from app import app,db
from model import User
#传入两个对象，一个是Flask的app对象，一个是SQLAlchemy
migrate= Migrate(app,db)
#实例化Manager类
manager= Manager(app)
#给manager添加一个db命令且传入一个MigrateCommand的类
manager.add_command('db',MigrateCommand)
（2）准备好数据模型。
（3）初始化和更新迁移数据库操作：
#用于初始化一个迁移脚本的环境
$ python manager.py db init
#表示创建迁移仓库
$ python manager.py db migrate
#更新数据库
$ python manager.py db upgrade
Flsak-Migrate数据库迁移实例：model.py
#encoding:utf8
from flask_sqlalchemy import SQLAlchemy
from app import db
class User(db.Model):
user_id = db.Column(db.Integer, primary_key=True)
user_name = db.Column(db.String(60), nullable=False)
user_password = db.Column(db.String(30), nullable=False)
新建文件app.py，内容如下：
Flsak-Migrate数据库迁移实例：app.py
# encoding:utf-8
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@127.0.0.1:3306/db_6_14'
# 'mysql+pymysql://root:root@127.0.0.1:3306/'
# 如果设置成 True (默认情况)，Flask-SQLAlchemy 将会追踪对象的修改并且发送信号。这需要额外的内存， 如果不必要的可以禁用它。
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)
@app.route('/')
def hello_world():
return 'Hello World!'
if __name__ == '__main__':
app.run()
Flsak-Migrate数据库迁移实例：manager.py
# encoding:utf8
from flask_script import Manager # flask 脚本
from flask_migrate import Migrate, MigrateCommand # flask 迁移数据
from app import app, db
from model import User
migrate = Migrate(app, db) # 传入2个对象一个是flask的app对象，一个是SQLAlchemy
manager = Manager(app)
# 给manager添加一个db命令并且传入一个MigrateCommand的类
manager.add_command('db', MigrateCommand)
if __name__ == '__main__':
manager.run()
requirements.txt
alembic==1.8.1
click==7.1.2
colorama==0.4.5
Flask==1.1.4
Flask-Migrate==2.7.0
Flask-Script==2.0.6
Flask-SQLAlchemy==2.5.1
greenlet==1.1.3
importlib-metadata==4.12.0
importlib-resources==5.9.0
itsdangerous==1.1.0
Jinja2==2.11.3
Mako==1.2.2
MarkupSafe==2.0.1
PyMySQL==1.0.2
SQLAlchemy==1.4.41
Werkzeug==1.0.1
zipp==3.8.1
接下来就可以进行初始化了，使用命令
$ python manager.py db init
该命令用于初始化一个迁移脚本的环境。这个命令中的’db’是在manager.add_command(‘db’,MigrateComand)这行代码中我们声明的命令行对象名称；
init是Migrate命令，表示初始化迁移数据库仓库，运行完成之后，会在当前工程目录下创建一个migrations文件夹，用于进行迁移的数据库脚本都放在这里。
接着使用命令便可以完成数据库的创建。
如果是对数据库中的表进行更新，同样也是使用下面的两条命令：
$ python manager.py db migrate
$ python manager.py db upgrade
在数据库中可以看到数据库已经成功创建了。
不使用Flask-Script，使用flask命令如下：
初始化数据库：flask db init
迁移新更改：flask db migrate
升级：flask db upgrade
注意：第一次迁移实际上相当于调用db.create_all()，但在后续迁移中，upgrade命令对表实施更新操作但不会影响表中的内容。
12.使用Python和 Flask设计RESTful APIs¶
近些年来 REST (REpresentational State Transfer) 已经变成了 web services 和 web APIs 的标配。
在本文中我将向你展示如何简单地使用 Python 和 Flask 框架来创建一个 RESTful 的 web service。
什么是 REST？
六条设计规范定义了一个 REST 系统的特点:
客户端-服务器: 客户端和服务器之间隔离，服务器提供服务，客户端进行消费。
无状态: 从客户端到服务器的每个请求都必须包含理解请求所必需的信息。换句话说， 服务器不会存储客户端上一次请求的信息用来给下一次使用。
可缓存: 服务器必须明示客户端请求能否缓存。
分层系统: 客户端和服务器之间的通信应该以一种标准的方式，就是中间层代替服务器做出响应的时候，客户端不需要做任何变动。
统一的接口: 服务器和客户端的通信方法必须是统一的。
按需编码: 服务器可以提供可执行代码或脚本，为客户端在它们的环境中执行。这个约束是唯一一个是可选的。
12.1 什么是一个 RESTful 的 web service？¶
REST 架构的最初目的是适应万维网的 HTTP 协议。
RESTful web services 概念的核心就是“资源”。 资源可以用 URI 来表示。客户端使用 HTTP 协议定义的方法来发送请求到这些 URIs，当然可能会导致这些被访问的”资源“状态的改变。
HTTP 标准的方法有如下:
========== ===================== ==================================
HTTP 方法 行为 示例
========== ===================== ==================================
GET 获取资源的信息 http://example.com/api/orders
GET 获取某个特定资源的信息 http://example.com/api/orders/123
POST 创建新资源 http://example.com/api/orders
PUT 更新资源 http://example.com/api/orders/123
DELETE 删除资源 http://example.com/api/orders/123
========== ====================== ==================================
REST 设计不需要特定的数据格式。在请求中数据可以以 JSON 形式, 或者有时候作为 url 中查询参数项。
12.2 设计一个简单的 web service¶
坚持 REST 的准则设计一个 web service 或者 API 的任务就变成一个标识资源被展示出来以及它们是怎样受不同的请求方法影响的练习。
比如说，我们要编写一个待办事项应用程序而且我们想要为它设计一个 web service。要做的第一件事情就是决定用什么样的根 URL 来访问该服务。例如，我们可以通过这个来访问:
http://[hostname]/todo/api/v1.0/
在这里我已经决定在 URL 中包含应用的名称以及 API 的版本号。在 URL 中包含应用名称有助于提供一个命名空间以便区分同一系统上的其它服务。在 URL 中包含版本号能够帮助以后的更新，如果新版本中存在新的和潜在不兼容的功能，可以不影响依赖于较旧的功能的应用程序。
下一步骤就是选择将由该服务暴露(展示)的资源。这是一个十分简单地应用，我们只有任务，因此在我们待办事项中唯一的资源就是任务。
我们的任务资源将要使用 HTTP 方法如下:
========== =============================================== =============================
HTTP 方法 URL 动作
========== =============================================== ==============================
GET http://[hostname]/todo/api/v1.0/tasks 检索任务列表
GET http://[hostname]/todo/api/v1.0/tasks/[task_id] 检索某个任务
POST http://[hostname]/todo/api/v1.0/tasks 创建新任务
PUT http://[hostname]/todo/api/v1.0/tasks/[task_id] 更新任务
DELETE http://[hostname]/todo/api/v1.0/tasks/[task_id] 删除任务
========== ================================================ =============================
我们定义的任务有如下一些属性:
id: 任务的唯一标识符。数字类型。
title: 简短的任务描述。字符串类型。
description: 具体的任务描述。文本类型。
done: 任务完成的状态。布尔值。
目前为止关于我们的 web service 的设计基本完成。剩下的事情就是实现它！
创建flask项目，flaskapp1
使用pychrm创建
使用命令行创建
$ mkdir flaskapp1
$ cd flaskapp1
$ virtualenv flask
New python executable in flask/bin/python
Installing setuptools............................done.
Installing pip...................done.
$ flask/bin/pip install flask
既然已经安装了 Flask，现在开始创建一个简单地网页应用，我们把它放在一个叫 app.py 的文件中:
#!flask/bin/python
from flask import Flask
app = Flask(__name__)
@app.route('/')
def index():
return "Hello, World!"
if __name__ == '__main__':
app.run(debug=True)
为了运行这个程序我们必须执行 app.py:
$ chmod a+x app.py
$ ./app.py
* Running on http://127.0.0.1:5000/
* Restarting with reloader
现在你可以启动你的网页浏览器，输入 http://localhost:5000 看看这个小应用程序的效果。
简单吧？现在我们将这个应用程序转换成我们的 RESTful service！
12.3 使用Python和 Flask实现RESTful services¶
在 Flask 中有许多扩展来帮助我们构建 RESTful services，但是在我看来这个任务十分简单，没有必要使用 Flask 扩展。
1.Get¶
我们现在来实现 web service 的第一个入口:
#!flask/bin/python
from flask import Flask, jsonify
app = Flask(__name__)
tasks = [
{
'id': 1,
'title': u'Buy groceries',
'description': u'Milk, Cheese, Pizza, Fruit, Tylenol',
'done': False
},
{
'id': 2,
'title': u'Learn Python',
'description': u'Need to find a good Python tutorial on the web',
'done': False
}
]
@app.route('/todo/api/v1.0/tasks', methods=['GET'])
def get_tasks():
return jsonify({'tasks': tasks})
if __name__ == '__main__':
app.run(debug=True)
正如你所见，没有多大的变化。我们创建一个任务的内存数据库，这里无非就是一个字典和数组。数组中的每一个元素都具有上述定义的任务的属性。
取代了首页，我们现在拥有一个 get_tasks 的函数，访问的 URI 为 /todo/api/v1.0/tasks，并且只允许 GET 的 HTTP 方法。
这个函数的响应不是文本，我们使用 JSON 数据格式来响应，Flask 的 jsonify 函数从我们的数据结构中生成。
使用网页浏览器来测试我们的 web service 不是一个最好的注意，因为网页浏览器上不能轻易地模拟所有的 HTTP 请求的方法。相反，我们会使用 curl。如果你还没有安装 curl 的话，请立即安装它。
通过执行 app.py，启动 web service。接着打开一个新的控制台窗口，运行以下命令:
$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 06:51:12 GMT
Content-Type: application/json
Content-Length: 310
Connection: close
{
"tasks": [
{
"description": "Milk, Cheese, Pizza, Fruit, Tylenol",
"done": false,
"id": 1,
"title": "Buy groceries"
},
{
"description": "Need to find a good Python tutorial on the web",
"done": false,
"id": 2,
"title": "Learn Python"
}
]
}
我们已经成功地调用我们的 RESTful service 的一个函数！
现在我们开始编写 GET 方法请求我们的任务资源的第二个版本。这是一个用来返回单独一个任务的函数:
from flask import abort
@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id):
task = [task for task in tasks if task['id'] == task_id]
# task = list(filter(lambda t: t['id'] == task_id, tasks))
if len(task) == 0:
abort(404)
return jsonify({'task': task[0]})
第二个函数有些意思。这里我们得到了 URL 中任务的 id，接着 Flask 把它转换成 函数中的 task_id 的参数。
我们用这个参数来搜索我们的任务数组。如果我们的数据库中不存在搜索的 id，我们将会返回一个类似 404 的错误，根据 HTTP 规范的意思是 “资源未找到”。
如果我们找到相应的任务，那么我们只需将它用 jsonify 打包成 JSON 格式并将其发送作为响应，就像我们以前那样处理整个任务集合。
调用 curl 请求的结果如下:
$ curl -i http://localhost:5000/todo/api/v1.0/tasks/1
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 06:53:30 GMT
Content-Type: application/json
Content-Length: 139
Connection: close
{
"task": {
"description": "Milk, Cheese, Pizza, Fruit, Tylenol",
"done": false,
"id": 1,
"title": "Buy groceries"
}
}
18793@DESKTOP-8LI950S ~
$ curl -i http://localhost:5000/todo/api/v1.0/tasks/2
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 06:53:36 GMT
Content-Type: application/json
Content-Length: 149
Connection: close
{
"task": {
"description": "Need to find a good Python tutorial on the web",
"done": false,
"id": 2,
"title": "Learn Python"
}
}
18793@DESKTOP-8LI950S ~
$ curl -i http://localhost:5000/todo/api/v1.0/tasks/3
HTTP/1.1 404 NOT FOUND
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 06:53:38 GMT
Content-Type: text/html; charset=utf-8
Content-Length: 207
Connection: close
<!doctype html>
<html lang=en>
<title>404 Not Found</title>
<h1>Not Found</h1>
<p>The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again.</p>
当我们请求 id #2 的资源时候，我们获取到了，但是当我们请求 #3 的时候返回了 404 错误。有关错误奇怪的是返回的是 HTML 信息而不是 JSON，这是因为 Flask 按照默认方式生成 404 响应。
由于这是一个 Web service 客户端希望我们总是以 JSON 格式回应，所以我们需要改善我们的 404 错误处理程序:
from flask import make_response
@app.errorhandler(404)
def not_found(error):
return make_response(jsonify({'error': 'Not found'}), 404)
我们会得到一个友好的错误提示:
$ curl -i http://localhost:5000/todo/api/v1.0/tasks/3
HTTP/1.1 404 NOT FOUND
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 06:57:35 GMT
Content-Type: application/json
Content-Length: 27
Connection: close
{
"error": "Not found"
}
2.Post¶
接下来就是 POST 方法，我们用来在我们的任务数据库中插入一个新的任务:
from flask import request
@app.route('/todo/api/v1.0/tasks', methods=['POST'])
def create_task():
if not request.json or not 'title' in request.json:
abort(400)
task = {
'id': tasks[-1]['id'] + 1,
'title': request.json['title'],
'description': request.json.get('description', ""),
'done': False
}
tasks.append(task)
return jsonify({'task': task}), 201
添加一个新的任务也是相当容易地。只有当请求以 JSON 格式形式，request.json 才会有请求的数据。如果没有数据，或者存在数据但是缺少 title 项，我们将会返回 400，这是表示请求无效。
接着我们会创建一个新的任务字典，使用最后一个任务的 id + 1 作为该任务的 id。我们允许 description 字段缺失，并且假设 done 字段设置成 False。
我们把新的任务添加到我们的任务数组中，并且把新添加的任务和状态 201 响应给客户端。
使用如下的 curl 命令来测试这个新的函数:
$ curl -i -H "Content-Type: application/json" -X POST -d '{"title":"Read a book"}' http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 201 CREATED
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 06:59:27 GMT
Content-Type: application/json
Content-Length: 102
Connection: close
{
"task": {
"description": "",
"done": false,
"id": 3,
"title": "Read a book"
}
}
18793@DESKTOP-8LI950S ~
$ curl -i http://localhost:5000/todo/api/v1.0/tasks/3
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 06:59:33 GMT
Content-Type: application/json
Content-Length: 102
Connection: close
{
"task": {
"description": "",
"done": false,
"id": 3,
"title": "Read a book"
}
}
注意：如果你在 Windows 上并且运行 Cygwin 版本的 curl，上面的命令不会有任何问题。然而，如果你使用原生的 curl，命令会有些不同:
$ curl -i -H "Content-Type: application/json" -X POST -d "{"""title""":"""Read a book"""}" http://localhost:5000/todo/api/v1.0/tasks
当然在完成这个请求后，我们可以得到任务的更新列表:
$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 07:02:22 GMT
Content-Type: application/json
Content-Length: 516
Connection: close
{
"tasks": [
{
"description": "Milk, Cheese, Pizza, Fruit, Tylenol",
"done": false,
"id": 1,
"title": "Buy groceries"
},
{
"description": "Need to find a good Python tutorial on the web",
"done": false,
"id": 2,
"title": "Learn Python"
},
{
"description": "",
"done": false,
"id": 3,
"title": "Read a book"
},
{
"description": "",
"done": false,
"id": 4,
"title": "Read a book"
}
]
}
3.PUT、Delete¶
剩下的两个函数如下所示:
@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
task = [task for task in tasks if task['id'] == task_id]
# task = list(filter(lambda t: t['id'] == task_id, tasks))
if len(task) == 0:
abort(404)
if not request.json:
abort(400)
if 'title' in request.json and type(request.json['title']) != "unicode":
abort(400)
if 'description' in request.json and type(request.json['description']) != "unicode":
abort(400)
if 'done' in request.json and type(request.json['done']) is not bool:
abort(400)
task[0]['title'] = request.json.get('title', task[0]['title'])
task[0]['description'] = request.json.get('description', task[0]['description'])
task[0]['done'] = request.json.get('done', task[0]['done'])
return jsonify({'task': task[0]})
@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
task = [task for task in tasks if task['id'] == task_id]
# task = list(filter(lambda t: t['id'] == task_id, tasks))
if len(task) == 0:
abort(404)
tasks.remove(task[0])
return jsonify({'result': True})
delete_task 函数没有什么特别的。
对于 update_task 函数，我们需要严格地检查输入的参数以防止可能的问题。
我们需要确保在我们把它更新到数据库之前，任何客户端提供我们的是预期的格式。
更新任务 #2 的函数调用如下所示:
$ curl -i -H "Content-Type: application/json" -X PUT -d '{"done":true}' http://localhost:5000/todo/api/v1.0/tasks/2
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 170
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 07:10:16 GMT
{
"task": [
{
"description": "Need to find a good Python tutorial on the web",
"done": true,
"id": 2,
"title": "Learn Python"
}
]
}
4.优化web service接口¶
当前我们还有一个问题，客户端有可能需要从返回的JSON中重新构造URI，如果将来加入新的特性时，可能需要修改客户端。（例如新增版本。）
我们可以返回整个URI的路径给客户端，而不是任务的id。为了这个功能，创建一个小函数生成一个“public”版本的任务URI返回：
from flask import url_for
def make_public_task(task):
new_task = {}
for field in task:
if field == 'id':
new_task['uri'] = url_for('get_task', task_id=task['id'], _external=True)
else:
new_task[field] = task[field]
return new_task
这里所有做的事情就是从我们数据库中取出任务并且创建一个新的任务，这个任务的 id 字段被替换成通过 Flask 的 url_for 生成的 uri 字段。
当我们返回所有的任务列表的时候，在发送到客户端之前通过这个函数进行处理:
@app.route('/todo/api/v1.0/tasks', methods=['GET'])
def get_tasks():
# return jsonify({'tasks': list(map(make_public_task, tasks))})
return jsonify({'tasks': [make_public_task(task) for task in tasks]})
这里就是客户端获取任务列表的时候得到的数据:
$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 07:26:33 GMT
Content-Type: application/json
Content-Length: 400
Connection: close
{
"tasks": [
{
"description": "Milk, Cheese, Pizza, Fruit, Tylenol",
"done": false,
"title": "Buy groceries",
"uri": "http://localhost:5000/todo/api/v1.0/tasks/1"
},
{
"description": "Need to find a good Python tutorial on the web",
"done": false,
"title": "Learn Python",
"uri": "http://localhost:5000/todo/api/v1.0/tasks/2"
}
]
}
这种办法避免了与其它功能的兼容，拿到的是完整uri而不是一个id。
5.加强 RESTful web service 的安全性¶
我们已经完成了我们 web service 的大部分功能，但是仍然有一个问题。我们的 web service 对任何人都是公开的，这并不是一个好主意。
我们有一个可以管理我们的待办事项完整的 web service，但在当前状态下的 web service 是开放给所有的客户端。 如果一个陌生人弄清我们的 API 是如何工作的，他或她可以编写一个客户端访问我们的 web service 并且毁坏我们的数据。
大部分初级的教程会忽略这个问题并且到此为止。在我看来这是一个很严重的问题，我必须指出。
确保我们的 web service 安全服务的最简单的方法是要求客户端提供一个用户名和密码。在常规的 web 应用程序会提供一个登录的表单用来认证，并且服务器会创建一个会话为登录的用户以后的操作使用，会话的 id 以 cookie 形式存储在客户端浏览器中。然而 REST 的规则之一就是 “无状态”， 因此我们必须要求客户端在每一次请求中提供认证的信息。
我们一直试着尽可能地坚持 HTTP 标准协议。既然我们需要实现认证我们需要在 HTTP 上下文中去完成，HTTP 协议提供了两种认证机制: Basic 和 Digest。
有一个小的 Flask 扩展能够帮助我们，我们可以先安装 Flask-HTTPAuth:
$ flask/bin/pip install flask-httpauth
比方说，我们希望我们的 web service 只让访问用户名 miguel 和密码 python 的客户端访问。 我们可以设置一个基本的 HTTP 验证如下:
from flask_httpauth import HTTPBasicAuth
auth = HTTPBasicAuth()
@auth.get_password
def get_password(username):
if username == 'admin':
return 'python'
return None
@auth.error_handler
def unauthorized():
return make_response(jsonify({'error': 'Unauthorized access'}), 401)
get_password 函数是一个回调函数，Flask-HTTPAuth 使用它来获取给定用户的密码。在一个更复杂的系统中，这个函数是需要检查一个用户数据库，但是在我们的例子中只有单一的用户因此没有必要。
error_handler 回调函数是用于给客户端发送未授权错误代码。像我们处理其它的错误代码，这里我们定制一个包含 JSON 数据格式而不是 HTML 的响应。
随着认证系统的建立，所剩下的就是把需要认证的函数添加 @auth.login_required 装饰器。例如:
@app.route('/todo/api/v1.0/tasks', methods=['GET'])
@auth.login_required
def get_tasks():
# return jsonify({'tasks': list(map(make_public_task, tasks))})
return jsonify({'tasks': [make_public_task(task) for task in tasks]})
如果现在要尝试使用 curl 调用这个函数我们会得到:
$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 401 UNAUTHORIZED
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 07:33:51 GMT
Content-Type: application/json
Content-Length: 37
WWW-Authenticate: Basic realm="Authentication Required"
Connection: close
{
"error": "Unauthorized access"
}
为了能够调用这个函数我们必须发送我们的认证凭据:
$ curl -u admin:python -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 07:34:16 GMT
Content-Type: application/json
Content-Length: 400
Connection: close
{
"tasks": [
{
"description": "Milk, Cheese, Pizza, Fruit, Tylenol",
"done": false,
"title": "Buy groceries",
"uri": "http://localhost:5000/todo/api/v1.0/tasks/1"
},
{
"description": "Need to find a good Python tutorial on the web",
"done": false,
"title": "Learn Python",
"uri": "http://localhost:5000/todo/api/v1.0/tasks/2"
}
]
}
认证扩展给予我们很大的自由选择哪些函数需要保护，哪些函数需要公开。
为了确保登录信息的安全应该使用 HTTP 安全服务器(例如: https://…)，这样客户端和服务器之间的通信都是加密的，以防止传输过程中第三方看到认证的凭据。
让人不舒服的是当请求收到一个 401 的错误，网页浏览都会跳出一个丑陋的登录框，即使请求是在后台发生的。因此如果我们要实现一个完美的 web 服务器的话，我们就需要禁止跳转到浏览器显示身份验证对话框，让我们的客户端应用程序自己处理登录。
一个简单的方式就是不返回 401 错误。403 错误是一个令人青睐的替代，403 错误表示 “禁止” 的错误:
@auth.error_handler
def unauthorized():
return make_response(jsonify({'error': 'Unauthorized access'}), 403)
6.可能的改进¶
我们编写的小型的 web service 还可以在不少的方面进行改进。
对于初学者来说，一个真正的 web service 需要一个真实的数据库进行支撑。我们现在使用的内存数据结构会有很多限制不应该被用于真正的应用。
另外一个可以提高的领域就是处理多用户。如果系统支持多用户的话，不同的客户端可以发送不同的认证凭证获取相应用户的任务列表。在这样一个系统中的话，我们需要第二个资源就是用户。
在用户资源上的 POST 的请求代表注册换一个新用户。
一个 GET 请求表示客户端获取一个用户的信息。
一个 PUT 请求表示更新用户信息，比如可能是更新邮箱地址。
一个 DELETE 请求表示删除用户账号。
GET 检索任务列表请求可以在几个方面进行扩展。首先可以携带一个可选的页的参数，以便客户端请求任务的一部分。另外，这种扩展更加有用：允许按照一定的标准筛选。
比如，用户只想要看到完成的任务，或者只想看到任务的标题以 A 字母开头。所有的这些都可以作为 URL 的一个参数项。
参考文献：
Designing a RESTful API with Python and Flask
https://www.cnblogs.com/vovlie/p/4178077.html
http://www.pythondoc.com/flask-restful/first.html
https://www.bookstack.cn/read/flask-restful/spilt.8.3b3bb30c584b4b80.md
13.使用Flask-RESTful设计RESTful API¶
前面我已经用 Flask 实现了一个 RESTful 服务器。今天我们将会使用 Flask-RESTful 来实现同一个 RESTful 服务器，Flask-RESTful 是一个可以简化 APIs 的构建的 Flask 扩展。
13.1 RESTful 服务器¶
作为一个提醒， 这里就是待完成事项列表 web service 所提供的方法的定义:
========== =============================================== =============================
HTTP 方法 URL 动作
========== =============================================== ==============================
GET http://[hostname]/todo/api/v1.0/tasks 检索任务列表
GET http://[hostname]/todo/api/v1.0/tasks/[task_id] 检索某个任务
POST http://[hostname]/todo/api/v1.0/tasks 创建新任务
PUT http://[hostname]/todo/api/v1.0/tasks/[task_id] 更新任务
DELETE http://[hostname]/todo/api/v1.0/tasks/[task_id] 删除任务
========== ================================================ =============================
这个服务唯一的资源叫做“任务”，它有如下一些属性:
id: 任务的唯一标识符。数字类型。
title: 简短的任务描述。字符串类型。
description: 具体的任务描述。文本类型。
done: 任务完成的状态。布尔值。
1. 安装flask-restful¶
$ pip install flask_restful
安装完成后就可以引入
from flask_restful import Api, Resource, reqparse, abort
13.2 路由¶
在上一遍文章中，我使用了 Flask 的视图函数来定义所有的路由。
Flask-RESTful 提供了一个 Resource 基础类，它能够定义一个给定 URL 的一个或者多个 HTTP 方法。
例如，定义一个可以使用 HTTP 的 GET, PUT 以及 DELETE 方法的 User 资源，你的代码可以如下:
from flask import Flask
from flask_restful import Api, Resource
app = Flask(__name__)
api = Api(app)
class UserAPI(Resource):
def get(self, id):
pass
def put(self, id):
pass
def delete(self, id):
pass
api.add_resource(UserAPI, '/users/<int:id>', endpoint='user')
add_resource 函数使用指定的 endpoint 注册路由到框架上。
如果没有指定 endpoint，Flask-RESTful 会根据类名生成一个，但是有时候有些函数比如 url_for 需要 endpoint，因此我会明确给 endpoint 赋值。
我的待办事项 API 定义两个 URLs：
/todo/api/v1.0/tasks（获取所有任务列表）
/todo/api/v1.0/tasks/（获取单个任务）。
我们现在需要两个资源:
from flask import Flask
from flask_restful import Api, Resource
app = Flask(__name__)
api = Api(app)
class TaskListAPI(Resource):
def get(self):
pass
def post(self):
pass
class TaskAPI(Resource):
def get(self, id):
pass
def put(self, id):
pass
def delete(self, id):
pass
api.add_resource(TaskListAPI, '/todo/api/v1.0/tasks', endpoint='tasks')
api.add_resource(TaskAPI, '/todo/api/v1.0/tasks/<int:id>', endpoint='task')
13.3 解析以及验证请求¶
当我在以前的文章中实现此服务器的时候，我自己对请求的数据进行验证。例如，在之前版本中如何处理 PUT 的:
@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
task = [task for task in tasks if task['id'] == task_id]
# task = list(filter(lambda t: t['id'] == task_id, tasks))
if len(task) == 0:
abort(404)
if not request.json:
abort(400)
if 'title' in request.json and type(request.json['title']) != "unicode":
abort(400)
if 'description' in request.json and type(request.json['description']) != "unicode":
abort(400)
if 'done' in request.json and type(request.json['done']) is not bool:
abort(400)
task[0]['title'] = request.json.get('title', task[0]['title'])
task[0]['description'] = request.json.get('description', task[0]['description'])
task[0]['done'] = request.json.get('done', task[0]['done'])
return jsonify({'task': task[0]})
在这里, 我必须确保请求中给出的数据在使用之前是有效，这样使得函数变得又臭又长。
Flask-RESTful 提供了一个更好的方式来处理数据验证，它叫做 RequestParser 类。
这个类工作方式类似命令行解析工具 argparse。
首先，对于每一个资源需要定义参数以及怎样验证它们:
from flask import Flask, request, jsonify, abort
from flask_restful import Api, Resource, reqparse
app = Flask(__name__)
api = Api(app)
tasks = [
{
'id': 1,
'title': u'Buy groceries',
'description': u'Milk, Cheese, Pizza, Fruit, Tylenol',
'done': False
},
{
'id': 2,
'title': u'Learn Python',
'description': u'Need to find a good Python tutorial on the web',
'done': False
}
]
from flask import url_for
def make_public_task(task):
new_task = {}
for field in task:
if field == 'id':
new_task['uri'] = url_for('get_task', task_id=task['id'], _external=True)
else:
new_task[field] = task[field]
return new_task
class TaskListAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title', type=str, required=True,
help='No task title provided', location='json')
self.reqparse.add_argument('description', type=str, default="", location='json')
super(TaskListAPI, self).__init__()
# 检索任务列表
def get(self):
pass
# 创建新任务
def post(self):
pass
class TaskAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title', type=str, location='json')
self.reqparse.add_argument('description', type=str, location='json')
self.reqparse.add_argument('done', type=bool, location='json')
super(TaskAPI, self).__init__()
# 检索某个任务
def get(self, id):
pass
# 更新任务
def put(self, id):
pass
# 删除任务
def delete(self, id):
pass
api.add_resource(TaskListAPI, '/todo/api/v1.0/tasks', endpoint='tasks')
api.add_resource(TaskAPI, '/todo/api/v1.0/tasks/<int:id>', endpoint='task')
if __name__ == '__main__':
app.run(host="0.0.0.0", debug=True)
在 TaskListAPI 资源中，POST 方法是唯一接收参数的。参数“标题”是必须的，因此我定义一个缺少“标题”的错误信息。当客户端缺少这个参数的时候，Flask-RESTful 将会把这个错误信息作为响应发送给客户端。
“描述”字段是可选的，当缺少这个字段的时候，默认的空字符串将会被使用。
一个有趣的方面就是 RequestParser 类默认情况下在 request.values 中查找参数，因此 location 可选参数必须被设置以表明请求过来的参数是 request.json 格式的。
TaskAPI 资源的参数处理是同样的方式，但是有少许不同。PUT 方法需要解析参数，并且这个方法的所有参数都是可选的。
使用 Flask-RESTful 来处理验证的另一个好处就是没有必要单独地处理类似 HTTP 400 错误，Flask-RESTful 会来处理这些。
13.4 生成响应¶
原来设计的 REST 服务器使用 Flask 的 jsonify 函数来生成响应。
Flask-RESTful 会自动地处理转换成 JSON 数据格式，因此下面的代码需要替换:
return jsonify({'tasks': [make_public_task(task) for task in tasks]})
现在需要写成这样:
return {'task': [make_public_task(task) for task in tasks]}
Flask-RESTful 也支持自定义状态码，如果有必要的话:
return {'task': [make_public_task(task) for task in tasks]}, 201
Flask-RESTful 还有更多的功能。make_public_task 能够把来自原始服务器上的任务从内部形式包装成客户端想要的外部形式。
最典型的就是把任务的 id 转成 uri。
Flask-RESTful 就提供一个辅助函数能够很优雅地做到这样的转换，不仅仅能够把 id 转成 uri 并且能够转换其他的参数:
from flask.ext.restful import fields, marshal
task_fields = {
'title': fields.String,
'description': fields.String,
'done': fields.Boolean,
'uri': fields.Url('task')
}
class TaskAPI(Resource):
# ...
def put(self, id):
# ...
return { 'task': marshal(task, task_fields) }
task_fields 结构用于作为 marshal 函数的模板。fields.Uri 是一个用于生成一个 URL 的特定的参数。 它需要的参数是 endpoint。
完整代码如下：
from flask import Flask, request, abort
from flask_restful import Api, Resource, reqparse, fields, marshal
from flask_httpauth import HTTPBasicAuth
auth = HTTPBasicAuth()
app = Flask(__name__)
api = Api(app)
task_fields = {
'title': fields.String,
'description': fields.String,
'done': fields.Boolean,
'uri': fields.Url('task')
}
tasks = [
{
'id': 1,
'title': u'Buy groceries',
'description': u'Milk, Cheese, Pizza, Fruit, Tylenol',
'done': False
},
{
'id': 2,
'title': u'Learn Python',
'description': u'Need to find a good Python tutorial on the web',
'done': False
}
]
class TaskListAPI(Resource):
# HTTP 基本身份验
# decorators = [auth.login_required]
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title', type=str, required=True,
help='No task title provided', location='json')
self.reqparse.add_argument('description', type=str, default="", location='json')
super(TaskListAPI, self).__init__()
# 检索任务列表
def get(self):
# return jsonify({'tasks': [make_public_task(task) for task in tasks]})
# 自定义状态码
# return {'task': marshal(tasks, task_fields)}, 201
return {'task': marshal(tasks, task_fields)}
# 创建新任务
def post(self):
if not request.json or not 'title' in request.json:
abort(400)
task = {
'id': tasks[-1]['id'] + 1,
'title': request.json['title'],
'description': request.json.get('description', ""),
'done': False
}
tasks.append(task)
return {'task': marshal(task, task_fields)}
class TaskAPI(Resource):
# HTTP 基本身份验
# decorators = [auth.login_required]
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title', type=str, location='json')
self.reqparse.add_argument('description', type=str, location='json')
self.reqparse.add_argument('done', type=bool, location='json')
super(TaskAPI, self).__init__()
# 检索某个任务
def get(self, id):
task = [task for task in tasks if task['id'] == id]
# task = list(filter(lambda t: t['id'] == task_id, tasks))
if len(task) == 0:
abort(404)
return {'task': marshal(task, task_fields)}
# 更新任务
def put(self, id):
task = [task for task in tasks if task['id'] == id]
# task = list(filter(lambda t: t['id'] == id, tasks))
if len(task) == 0:
abort(404)
task = task[0]
# print(task)
args = self.reqparse.parse_args()
for k, v in list(args.items()):
if v is not None:
task[k] = v
return {'task': marshal(task, task_fields)}
# return jsonify({'task': make_public_task(task)})
# 删除任务
def delete(self, id):
task = [task for task in tasks if task['id'] == id]
# task = list(filter(lambda t: t['id'] == id, tasks))
if len(task) == 0:
abort(404)
tasks.remove(task[0])
return {'result': True}
api.add_resource(TaskListAPI, '/todo/api/v1.0/tasks', endpoint='tasks')
api.add_resource(TaskAPI, '/todo/api/v1.0/tasks/<int:id>', endpoint='task')
if __name__ == '__main__':
app.run(host="0.0.0.0", debug=True)
调用 curl 请求的结果如下:
# 检索任务列表
$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 08:56:25 GMT
Content-Type: application/json
Content-Length: 425
Connection: close
{
"task": [
{
"title": "Buy groceries",
"description": "Milk, Cheese, Pizza, Fruit, Tylenol",
"done": false,
"uri": "/todo/api/v1.0/tasks/1"
},
{
"title": "Learn Python",
"description": "Need to find a good Python tutorial on the web",
"done": false,
"uri": "/todo/api/v1.0/tasks/2"
}
]
}
# 创建新任务
$ curl -i -H "Content-Type: application/json" -X POST -d '{"title":"Read a book"}' http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 08:56:44 GMT
Content-Type: application/json
Content-Length: 146
Connection: close
{
"task": {
"title": "Read a book",
"description": "",
"done": false,
"uri": "/todo/api/v1.0/tasks/3"
}
}
# 检索某个任务
$ curl -i http://localhost:5000/todo/api/v1.0/tasks/1
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 08:57:03 GMT
Content-Type: application/json
Content-Length: 219
Connection: close
{
"task": [
{
"title": "Buy groceries",
"description": "Milk, Cheese, Pizza, Fruit, Tylenol",
"done": false,
"uri": "/todo/api/v1.0/tasks/1"
}
]
}
# 更新任务
$ curl -i -H "Content-Type: application/json" -X PUT -d '{"done":true}' http://localhost:5000/todo/api/v1.0/tasks/2
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 09:15:12 GMT
Content-Type: application/json
Content-Length: 192
Connection: close
{
"task": {
"title": "Learn Python",
"description": "Need to find a good Python tutorial on the web",
"done": true,
"uri": "/todo/api/v1.0/tasks/2"
}
}
# 删除任务
$ curl -i -X DELETE http://localhost:5000/todo/api/v1.0/tasks/3
HTTP/1.1 200 OK
Server: Werkzeug/2.2.2 Python/3.8.8
Date: Mon, 05 Sep 2022 09:17:30 GMT
Content-Type: application/json
Content-Length: 23
Connection: close
{
"result": true
}
13.5 flask+mysql数据库CURD案例¶
参考demo案例
https://gitee.com/hujianli94net/flask-restful-api-template
参考demo
https://github.com/tomasrasymas/flask-restful-api-template
https://github.com/tomasrasymas/flask-restful-api-template/blob/master/endpoints/todos/resource.py
13.6 认证¶
在 REST 服务器中的路由都是由 HTTP 基本身份验证保护着。在最初的那个服务器是通过使用 Flask-HTTPAuth 扩展来实现的。
因为 Resouce 类是继承自 Flask 的 MethodView，它能够通过定义 decorators 变量并且把装饰器赋予给它:
from flask.ext.httpauth import HTTPBasicAuth
# ...
auth = HTTPBasicAuth()
# ...
class TaskAPI(Resource):
# HTTP 基本身份验
decorators = [auth.login_required]
# ...
class TaskAPI(Resource):
# HTTP 基本身份验
decorators = [auth.login_required]
# ...
13.7 在flask-restful中添加日志¶
在flask-restful中，logger的使用有更优雅的方式，来看示例
from flask import Flask, request, abort
from flask_restful import Api, Resource, reqparse, fields, marshal
from flask_httpauth import HTTPBasicAuth
import logging.config
logging.config.dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"simple": {"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "DEBUG",
"formatter": "simple",
"stream": "ext://sys.stdout",
},
"info_file_handler": {
"class": "logging.handlers.RotatingFileHandler",
"level": "INFO",
"formatter": "simple",
"filename": "info.log",
"maxBytes": 10485760,
"backupCount": 50,
"encoding": "utf8",
},
"error_file_handler": {
"class": "logging.handlers.RotatingFileHandler",
"level": "ERROR",
"formatter": "simple",
"filename": "errors.log",
"maxBytes": 10485760,
"backupCount": 20,
"encoding": "utf8",
},
"debug_file_handler": {
"class": "logging.handlers.RotatingFileHandler",
"level": "DEBUG",
"formatter": "simple",
"filename": "debug.log",
"maxBytes": 10485760,
"backupCount": 50,
"encoding": "utf8",
},
},
"loggers": {
"my_module": {"level": "ERROR", "handlers": ["console"], "propagate": "no"}
},
"root": {
"level": "DEBUG",
"handlers": ["error_file_handler", "debug_file_handler"],
},
}
)
auth = HTTPBasicAuth()
app = Flask(__name__)
api = Api(app, default_mediatype="application/json")
tasks = [
{
'id':1,
'title':'白鹿原',
'description':'这是一定死啦塞啊；阿凯骄傲灵丹安矿鉴定暗访东方三；啊反馈；安静快撒娇扥卡机',
'done':False
},
{
'id':2,
'title':'嫌疑人',
'description':'as大肯借；额看课文静安寺放大反抗航将；来否；付定金奥兰多 看企鹅长恨东澳岛啊；伐开森',
'done':False
}
]
task_fields = {
'title': fields.String,
'description': fields.String,
'done': fields.Boolean,
'uri': fields.Url('task')
}
class TaskListAPI(Resource):
# HTTP 基本身份验
# decorators = [auth.login_required]
def __init__(self, **kargs):
self.logger = kargs.get('logger')
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title', type=str, required=True, help='No task title provided', location='json')
self.reqparse.add_argument('description', type=str, default="", location='json')
self.logger.debug(self.reqparse)
super(TaskListAPI, self).__init__()
# 检索任务列表
def get(self):
# return jsonify({'tasks': [make_public_task(task) for task in tasks]})
# 自定义状态码
# return {'task': marshal(tasks, task_fields)}, 201
return {'task': marshal(tasks, task_fields)}
# 创建新任务
def post(self):
if not request.json or not 'title' in request.json:
abort(400)
task = {
'id': tasks[-1]['id'] + 1,
'title': request.json['title'],
'description': request.json.get('description', ""),
'done': False
}
tasks.append(task)
return {'task': marshal(task, task_fields)}
class TaskAPI(Resource):
# HTTP 基本身份验
# decorators = [auth.login_required]
def __init__(self, **kargs):
self.logger = kargs.get('logger')
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('title', type=str, location='json')
self.reqparse.add_argument('description', type=str, location='json')
self.reqparse.add_argument('done', type=bool, location='json')
self.logger.debug(self.reqparse)
super(TaskAPI, self).__init__()
# 检索某个任务
def get(self, id):
task = [task for task in tasks if task['id'] == id]
# task = list(filter(lambda t: t['id'] == task_id, tasks))
if len(task) == 0:
abort(404)
return {'task': marshal(task, task_fields)}
# 更新任务
def put(self, id):
task = [task for task in tasks if task['id'] == id]
# task = list(filter(lambda t: t['id'] == id, tasks))
if len(task) == 0:
abort(404)
task = task[0]
# print(task)
args = self.reqparse.parse_args()
for k, v in list(args.items()):
if v is not None:
task[k] = v
return {'task': marshal(task, task_fields)}
# return jsonify({'task': make_public_task(task)})
# 删除任务
def delete(self, id):
task = [task for task in tasks if task['id'] == id]
# task = list(filter(lambda t: t['id'] == id, tasks))
if len(task) == 0:
abort(404)
tasks.remove(task[0])
return {'result': True}
api.add_resource(TaskListAPI, '/todo/api/v1.0/tasks', endpoint='tasks',
resource_class_kwargs={"logger": logging.getLogger('/TaskListAPI')})
api.add_resource(TaskAPI, '/todo/api/v1.0/tasks/<int:id>', endpoint='task',
resource_class_kwargs={"logger": logging.getLogger('/TaskAPI')})
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, use_reloader=True,debug=True)
我们使用上次用到的dictConfig，
主要的区别在于api.add_resource()方法中，使用了参数resource_class_kwargs，然后在Resource子类中的构造函数__init__，将日志记录器获取到，后面就可以在各个处理方法中使用了。
再次使用postman发起POST请求，可以看到debug.log是这个样子的
参考文献：
13.8 使用 Flask 设计 RESTful 的认证¶
http://www.pythondoc.com/flask-restful/third.html
https://www.jianshu.com/p/2bbf15db8dda
使用Flask设计带认证token的RESTful API接口
https://www.cnblogs.com/vovlie/p/4182814.html
参考demo
https://www.bookstack.cn/read/flask-restful/spilt.1.dc04892e18a3e9e6.md
14.Flask + PyJWT 实现基于JWT的用户认证授权¶
15.跨域访问¶
15.1 什么是跨域？¶
跨域是指，浏览器从服务器A获取的静态资源，包括html、css、javascript，然后在javascript中通过ajax访问服务器B的静态资源或请求。
15.2 CORS¶
w3c组织制定了
[Cross Origin Resource Sharing](https://www.w3.org/TR/cors/)
的规范，简写为CORS，现在这个规范已经被大多数浏览器支持。
使用上面的示例
前端页面index.html
<html>
<body>
<button type="button" onclick="jump()">Click Me!</button>
<script>
function jump(){
let xhr = new XMLHttpRequest();
xhr.open('GET', "http://172.16.3.134:5000/todo/api/v1.0/tasks", true);
xhr.send();
xhr.onreadystatechange = processRequest;
function processRequest(e) {
if (xhr.readyState == 4 && xhr.status == 200) {
let response = JSON.parse(xhr.responseText);
console.log(response)
}
}
}
</script>
</body>
</html>
我们将index.html部署在一台机器上(172.16.3.134)，flask应用部署在另一台机器上(172.16.3.134)，然后在浏览器中访问index.html，点击页面中的按钮，这时候就会报错了
15.3 Flask配置cors¶
CORS需要在后端应用中进行配置。在flask中，可以使用扩展flask-cors，首先安装
$ pip install flask-cors
接下来来到app.py，导入模块，并将flask应用包括起来就可以了，如下
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
重新启动应用，再次访问index.html，这时候，返回的结果就正常了
16.网站部署¶
Flask自带的Web服务器是轻量级服务器，以方便程序员开发和调试。但是在实际部署的时候，却不能直接用Flask发布应用，因为Flask自带的Web服务器还不够强壮，无法在生产环境中使用，还需要Web服务和WSGI（Web服务网关接口）。
本章主要涉及的知识点有：
Nginx+Gunicorn+Flask服务器的部署方法。
需求文件requirements.txt的创建和使用方法。
16.1 服务器部署¶
服务器部署方案主要有Nginx+Gunicorn+Flask或Nginx+uwsgi+Flask两种方案。
由于Nginx+uwsgi+Flask方案可能存在内存泄漏（内存占用无限增加）以及uwsgi中设置配置文件比较烦琐等问题，因此推荐使用Nginx+Gunicorn+Flask方案。
这里使用Gunicorn作为应用服务器，Nginx作为反向代理服务器。
常用的WSGI容器有Gunicorn和uWSGI，但Gunicorn直接用命令启动，不需要编写配置文件，相对uWSGI要容易很多，所以这里推荐使用Gunicorn作为容器。
参考文献
https://www.cnblogs.com/Ray-liang/p/4837850.html
How To Serve Flask Applications with Gunicorn and Nginx on Ubuntu 20.04