目录
flask 三剑客
from flask import Flask,render_template,redirect,jsonify
app=Flask(__name__)
app.debug = True
@app.route('/')
def index():
#
# #1直接返回字符串
# return "ok1"
# #2返回html
# # name_dict={"name":"jason-gdx"}
# # return render_template("index.html",name="jason-gdx",name_dict=name_dict)
# #3跳转页面
# return "hADSG"
return redirect("/login")
# #返回json数据
# name_dict = [{'name': "jason-gdx"},{'name': "tank-sb"}]
# return jsonify(name_dict)
@app.route('/login')
def login():
return "ojbk"
if __name__ == '__main__':
# from werkzeug.serving import run_simple
# 内部run方法调用 run_simple(host, port, self, **options)，self就是app对象，当启动服务，
#self默认加括号调用，触发Flask类内部__call__的执行,__call__内部
# return self.wsgi_app(environ, start_response) flask的源码部分就在这里了。
app.run()
flask的配置文件
from flask import Flask
app = Flask(__name__)
#配置文件1,这中方式只能配置两种
# app.debug=True
# app.secret_key="123123"
#第二种，以字典的形式
# app.config['DEBUG']=True
#第三种,以文件的形式
app.config.from_pyfile("settings.py")
'''
# 同一目录下的settings文件
DEBUG = True
'''
#第四种以类的形式(推荐)
app.config.from_object('settings.DevelopmentConfig')
@app.route('/login/')
def login():
print('aaaa')
return 'ok'
if __name__ == '__main__':
app.run()
settings.py
# 公共配置
class Config(object):
DEBUG = False
TESTING = False
DATABASE_URI = 'sqlite://:memory:'
# 继承Config，在启动文件已拼接字符串的格式找到当前文件下的这个类，就可以切换到上线环境的配置
class ProductionConfig(Config):
DATABASE_URI = 'mysql://user@localhost/foo'
# 继承Config，在启动文件已拼接字符串的格式找到当前文件下的这个类，就可以切换到开发环境的配置
class DevelopmentConfig(Config):
DEBUG = True
# 测试环境
class TestingConfig(Config):
TESTING = True
路由本质
from flask import Flask,url_for,redirect
app = Flask(__name__)
# app.debug=True
# 方式1：
@app.route('/login', endpoint='sb') # app.route('/login/') 返回decorator，
# 相当于@decorator ==> login = decorator(login)
def login():
print('aaaa')
return "okoooo"
# 方式2：
# app.add_url_rule("/login", view_func=login, endpoint='sb') # 相当于在视图函数上面的语法糖@app.route('/login',endpoint='sb')
@app.route('/')
def index():
# real_url = url_for('login') # endpoint别名默认是函数名
real_url = url_for('sb') # 根据别名反向解析起别名的函数的url，没传endpoint，url_for使用视图函数名反向解析url
print(real_url) # /login
return redirect(real_url)
if __name__ == '__main__':
app.run()
'''
在decorator函数（闭包）内部调用了函数self.add_url_rule(rule, endpoint, f, **options)
rule： 路由 /login
endpoint: 给视图函数起别名，在另一个视图使用url_for就能根
据这个别名解析出其路由，进行跳转（redirect）,也可以不起别名，默认是视图函数名（view_func.__name__）
f: 视图函数 def login
options： 额外参数（比如在@app.route()中添加关键字参数endpoint，或别的关键字参数）
所以可以将@app.route改为 ：
app.add_url_rule('/login',endpoint,'login' **options)
'''
请求方式
from flask import Flask,url_for,redirect
app = Flask(__name__)
@app.route('/login', endpoint='sb', methods=['POST','GET' ]) # 以可迭代对象的形式，写上允许的请求方式，大小写均可。 不填methods默认是GET请求，具体看源码。
def login():
print('aaaa')
return "ok!"
路由的有名无名分组
@app.route('/login/<int:id>', endpoint='sb', methods = ['POST','GET']) # app.route('/login/') 返回decorator，
# 相当于@decorator ==> login = decorator(login)
def login(id):
print(id,type(id)) # 100 <class 'int'>
return "okoooo"
'''
可以限制路由分组的类型：包括以下几种
'default','string','any(任意类型)','path','int','float','uuid'
'''
flask 中视图类
方式1：继承views.View
from flask import Flask,url_for,views
app=Flask(__name__)
app.debug=True
def auth(view): # view = index
print('auth函数')
return view
class IndexView(views.View):
methods=["POST","GET"] # 允许的请求方法
decorators=[auth,]
def dispatch_request(self):
print('Index')
return 'Index!'
app.add_url_rule('/index', view_func= IndexView.as_view(name='index'), endpoint='sb')
'''
IndexView.as_view(name='index')调用views.View里面的as_view返回view
这个view就是我们起的别名name = index ,此时view_func=index,调用as_view()中的
闭包函数，返回（return self.dispatch_request(*args, **kwargs)）
所以在我们的视图类中要重写dispatch_request方法，进行请求分发
为什么要as_view(name='index')，给闭包起了个别名，避免冲突。
如果不指定，就都是view，所以是必填参数
在as_view源码中首先是从我们的视图类中获取cls.decorator列表,然后for循环加括号执行这些函数，所以还
可以给dispatch_request加多个装饰器函数，之后再调用
'''
# 依然还可以给view_func起个别名（endpoint='sb'），就是给view_func=view,也就
# 是给view起的别名name=index再起别名，在别的视图函数中反向解析出其url
@app.route('/')
def wrap():
real_url = url_for('sb')
print(real_url) # /index
return 'wrap视图函数'
if __name__ == '__main__':
app.run()
方式2：继承views.MethodView
直接反射我们视图函数内写的请求方法
from flask import Flask,url_for,views
app=Flask(__name__)
app.debug=True
class IndexView(views.MethodView):
# methods=["GET"]
# decorators=[,,,,]
def get(self):
print('Index')
return 'Index!,get'
def post(self):
return 'Index!,post'
app.add_url_rule('/index', view_func= IndexView.as_view(name='index'), endpoint='sb')
@app.route('/')
def wrap():
real_url = url_for('sb')
print(real_url) # /index
return 'wrap视图函数'
if __name__ == '__main__':
app.run()
前后端页面搭建小李子
from flask import Flask,render_template,request,redirect,session,url_for
app = Flask(__name__)
app.debug = True
app.secret_key = 'sdfsdfsdfsdf'
USERS = {
1:{'name':'张三','age':18,'gender':'男','text':"最帅"},
2:{'name':'李四','age':28,'gender':'男','text':"最阳关"},
3:{'name':'王五','age':18,'gender':'女','text':"最土.......豪"},
}
@app.route('/detail/<int:nid>',methods=['GET'])
def detail(nid):
info = USERS.get(nid)
return render_template('detail.html',info=info)
@app.route('/index',methods=['GET'])
def index():
return render_template('index.html',user_dict=USERS)
def login():
if request.method == "GET":
return render_template('login.html')
else:
# request.query_string
user = request.form.get('user')
pwd = request.form.get('pwd')
if user == 'cxw' and pwd == '123':
url=url_for('index')
return redirect(url)
# session['user_info'] = user
return render_template('login.html', error='用户名或密码错误')
app.add_url_rule("/login",view_func=login,methods=["GET","POST"])
if __name__ == '__main__':
app.run()
所需的html页面需要在放在根目录的templates文件夹内部，这个文件夹需要我们自己建。内部会自动调取所需的html页面。
index.html
<!DOCTYPE html>
<html lang="en">
<head>
<script src="https://cdn.bootcss.com/jquery/3.4.1/jquery.min.js"></script>
<script src="https://cdn.bootcss.com/twitter-bootstrap/4.4.1/js/bootstrap.min.js"></script>
<link href="https://cdn.bootcss.com/twitter-bootstrap/4.4.1/css/bootstrap.min.css" rel="stylesheet">
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
user_dict
<h1>用户列表</h1>
<table>
<thead>
<tr>
<th>序号</th>
<th>姓名</th>
<th>年龄</th>
<th>性别</th>
<th>备注</th>
</tr>
</thead>
<tbody>
{% for k,v in user_dict.items() %}
<tr>
<td>{{ k }}</td>
<td>{{ v.name }}</td>
<td>{{ v['age'] }}</td>
<td>{{ v.get('gender') }}</td>
<td><a href="/detail/{{k}}">查看详情</a></td>
</tr>
{% endfor %}
</tbody>
</table>
</body>
</html>
detail.html
<!DOCTYPE html>
<html lang="en">
<head>
<script src="https://cdn.bootcss.com/jquery/3.4.1/jquery.min.js"></script>
<script src="https://cdn.bootcss.com/twitter-bootstrap/4.4.1/js/bootstrap.min.js"></script>
<link href="https://cdn.bootcss.com/twitter-bootstrap/4.4.1/css/bootstrap.min.css" rel="stylesheet">
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
user_dict
<h1>用户列表</h1>
<table>
<td>{{ info.text }}</td>
</table>
</body>
</html>
login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="" method="post">
<h1>登录页面</h1>
姓名：<input type="text" name = 'user'>
密码：<input type="password" name="pwd">
<input type="submit" value="登录">
</form>
</body>
</html>
url的正则匹配
from flask import Flask, views, url_for
from werkzeug.routing import BaseConverter
app = Flask(import_name=__name__)
class RegexConverter(BaseConverter):
"""
自定义URL匹配正则表达式
"""
def __init__(self, map, regex):
super(RegexConverter, self).__init__(map)
self.regex = regex
print('map:',map) # map: Map([<Rule '/static/<filename>' (HEAD, OPTIONS, GET) -> static>])
print('regex:',regex) # regex: d+
def to_python(self, value):
"""
路由匹配时，匹配成功后传递给视图函数中参数的值
"""
#value就正则匹配出来的结果
print('value',value,type(value)) # value 777 <class 'str'>
return "asdasdasd"
def to_url(self, value):
"""
使用url_for反向生成URL时，传递的参数经过该方法处理，返回的值用于生成URL中的参数
"""
print('value1',value) # value1 999
val = super(RegexConverter, self).to_url(value)
print('to_url:',val) # to_url: 999
return '888'
# 添加到flask中
# 我们要用自定义的路由，用正则的话
#1导入from werkzeug.routing import BaseConverter
# 2我先要写一个类，然后继承BaseConverter，然后实现__inti__, def to_python(self, value):to_url(self, value)
# 3 app.url_map.converters['别名'] = RegexConverter
#4 我们在路由里面@app.route('/index/<regex1("d+"):nid>')，regex1=别名，regex1("正则表达式")
#5 regex1("正则表达式")匹配出来的结果，给to_python，一定要return
#6 当我们做反向解析的解析的时候，我们的参数，会传递给to_url，return的结果才是我们拼接到我们路由上
app.url_map.converters['regex1'] = RegexConverter
@app.route('/index/<regex1("d+"):nid>',endpoint="sb") # 将url中的字符串按照我们的正则匹配出来
def index(nid): # to_python返回的结果
print("nid",nid,type(nid)) # nid asdasdasd <class 'str'>
print(url_for('sb', nid='999')) # /index/888 to_url返回的结果
return 'Index'
if __name__ == '__main__':
app.run()
模板语言
比django的DTL（django templates language强大），可以加括号执行函数，也可以给函数传参，参数个数不受限制，而django只能传一个。
from flask import Flask,render_template,Markup,jsonify,make_response
app = Flask(__name__)
def func1(arg):
return Markup("<input type='text' value='%s' />" %(arg,))
@app.route('/')
def index():
return render_template('index.html',ff = func1, html=<h1>hello world!</h>)
if __name__ == '__main__':
app.run()
templates/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<!--同django一样，加个safe-->
{{html|safe}}
{{ff('六五')}}
{{ff('六五')|safe}}
</body>
</html>
注意：
1.Markup等价于django的mark_safe,
2.extends,include一模一样。
请求与响应
from flask import Flask
from flask import request # 这个request是全局的，任何地方都能获取到
from flask import render_template
from flask import redirect
from flask import make_response
from flask import jsonify # 返回json格式数据
app = Flask(__name__)
@app.route('/login.html', methods=['GET', "POST"])
def login():
# print(request.path) # /login.html
# 请求相关信息
# request.method 提交的方法
# request.args get请求提及的数据
# request.form post请求提交的数据
# request.values post和get提交的数据总和
# request.cookies 客户端所带的cookie
# request.headers 请求头
# request.path 不带域名，请求路径
# request.full_path 不带域名，带参数的请求路径
# request.script_root
# request.url 带域名带参数的请求路径
# request.base_url 带域名请求路径
# request.url_root 域名
# request.host_url 域名
# request.host 127.0.0.1:500
# request.files
# obj = request.files['the_file_name']
# obj.save('/var/www/uploads/' + secure_filename(f.filename))
# request.getdata contentType编码格式是application/json的数据 stringfy
# 响应相关信息
# return "字符串"
# return render_template('html模板路径',**{})
# return redirect('/index.html')
# return jsonify({'k1':'v1'})
response = make_response(render_template('index.html'))
# response是flask.wrappers.Response类型
response.delete_cookie('key') # 删除客户端cookie key
response.set_cookie('name', 'zhang') # 将cookie设置到客户端浏览器
response.headers['X-Something'] = 'A value' # 将响应头设置到客户端浏览器
# 客户端的设置跟响应的结果无关，即使不返回response也一样进行
# 上面的操作将cookie和header带到客户端浏览器进行设置或者删除cookie。
# return response
return "内容"
if __name__ == '__main__':
app.run()
session
cookie : 存放在客户端的键值对
session ： 存放在客户端的键值对
token : 存放在客户端，通过算法来校验
在使用session之前必须先设置一下密钥
app.secret_key = 'abc' # 值随便设 但不能不设置，不然启动报错
除请求对象之外，还有一个session对象。它允许你在不同请求间存储特定用户的信息。它是在Cookie的基础上实现的，并且对Cookie进行密钥签名要使用会话，你需要设置一个密钥。（app.session_interface对象）
设置：session['username'] ＝ 'xxx'
'''
在django中发生三件事，1，生成一个随机的字符串 2 往数据库存 3 写入cookie返回浏览器
在flask中他没有数据库，但session是怎样实现的？
生成一个密钥写入这个cookie,然后下次请求的时候，通过这个cookie解密，然后赋值给session
我们通过app.session_interface来查看
'''
删除：session.pop('username', None)
app.session_interface中save_session的参数（设置cookie的参数）
# 这些参数从
key, 键
value='', 值
max_age=None, 超时时间 cookie需要延续的时间（以秒为单位）如果参数是 None`` ，这个cookie会延续到浏览器关闭为止
expires=None, 超时时间(IE requires expires, so set it if hasn't been already.)
path='/', Cookie生效的路径，/ 表示根路径，特殊的：根路径的cookie可以被任何url的页面访问，浏览器只会把cookie回传给带有该路径的页面，这样可以避免将cookie传给站点中的其他的应用。
domain=None, Cookie生效的域名 你可用这个参数来构造一个跨站cookie。如， domain=".example.com"所构造的cookie对下面这些站点都是可读的：www.example.com 、 www2.example.com 和an.other.sub.domain.example.com 。如果该参数设置为 None ，cookie只能由设置它的站点读取
secure=False, 浏览器将通过HTTPS来回传cookie
httponly=False 只能http协议传输，无法被JavaScript获取（不是绝对，底层抓包可以获取到也可以被覆盖）
小李子：
from flask import Flask,session
app = Flask(__name__)
app.secret_key="abc"
app.config['SESSION_COOKIE_NAME']="dsb"
# app.session_interface
'''
app.session_interface这里面看
存session：
1 调用save_session,将我们的session加密的val,读取配置文件['SESSION_COOKIE_NAME']得到key
2 将1中的key,val存储到cookies（保存到客户端浏览器）
取session
1 获取request里面的cookies,获取里面key,这个key就是['SESSION_COOKIE_NAME']，值就是加密的值
2 对该值进行解密
'''
@app.route("/")
def index():
session['username']="zhang" # 调用这个url，将session保存到客户端浏览器
return "ok"
@app.route("/index1")
def index1():
print(session['username']) # zhang 调用这个url，将客户端的cookie携带到服务端，源码通过request.cookie.get('username')获取并保存到session。
return "ok1"
if __name__ == '__main__':
app.run()
闪现
'''
什么是闪现
a 产生信息，传给 c 页面
但是用户访问a 页面以后，不是直接跳转到c,而是到b,或则是其他页面，但是用户访问c页面的时候，我希望把a给我的信息拿到
'''
from flask import Flask,flash,get_flashed_messages,request
app = Flask(__name__)
app.secret_key = 'abc'
'''
flash：存信息
get_flashed_messages： 取信息
1. 如果要用flash就必须设置app.secret_key = 'abc'
2. 只能取一次，不重启项目，再次发送请求就取不到了
3. 我们可以通过 flash('普通信息',category="info")，对信息做分类
4. get_flashed_messages(with_categories=True,category_filter=("error",))，with_categories以键值对的形式获取
我们设置闪现，category_filter=("error",)进行分类信息的过滤
'''
@app.route('/index')
def index():
#(category="message", message))
flash('超时错误',category="error")
flash('普通信息',category="info")
return "abc"
# return redirect('/error')
@app.route('/error1')
def error1():
return "ok"
@app.route('/error')
def error():
# data = get_flashed_messages(category_filter=("error","info"))
# data1 = get_flashed_messages(category_filter=("error", "info"))
# print("data1",data1) # # data1 ['超时错误', '普通信息']
# print("data",data) # # data ['超时错误', '普通信息']
data = get_flashed_messages(with_categories=True, category_filter=("error",))
data1 = get_flashed_messages(with_categories=True, category_filter=("info"))
data2 = get_flashed_messages(with_categories=True, category_filter=("error", "info"))
print("data", data) # data [('error', '超时错误')]
print("data1", data1) # data1 [('info', '普通信息')]
print("data2", data2) # data2 [('error', '超时错误'), ('info', '普通信息')]
return "错误信息"
if __name__ == '__main__':
app.run()
请求扩展
1. before_request和after_request
from flask import Flask,request,render_template
app = Flask(__name__)
@app.before_request
def before1():
print(request)
print("我是请求之前1")
return "123"
@app.before_request
def before2():
print("我是请求之前2")
@app.after_request
def after1(response):
print("我是请求之后1")
return response
@app.after_request
def after2(response):
print("我是请求之后2")
return response # '123'
@app.route('/index')
def index():
print("我是真的视图")
return render_template("index.html")
if __name__ == '__main__':
# app.__call__
app.run()
2. before_first_request
from flask import Flask,request,render_template
app = Flask(__name__)
'''
Flask全局设置了一个属性self._got_first_request = False，当
第一次请求来时执行before_first,并把_got_first_request设置
成True，之后浏览器再对 /index 接口发送请求就不再调用before_first
函数了，直接调用 /index 接口，具体看源码：
app.__call__ ->wsgi_app() -> full_dispatch_request() ->
try_trigger_before_first_request_functions()
'''
@app.before_first_request
def before_first():
print("123")
@app.route('/index')
def index():
print("我是真的视图")
return render_template("index.html")
if __name__ == '__main__':
# app.__call__
app.run()
3. teardown_request
from flask import Flask,request,render_template
app = Flask(__name__)
# 如论有没有异常都会执行，如果没有异常这个参数就是None，有就记录这个异常
@app.teardown_request
def tear(e):
print('teardown_request')
print(e)
@app.route('/index')
def index():
# try:
# abc
# return render_template("index.html")
# except Exception as e:
# import traceback
# print(traceback.format_exc())
# return '123'
print('index视图函数')
abc
if __name__ == '__main__':
# app.__call__
app.run()
4. app.errorhandler
能捕捉到异常，客户端访问时能得到响应的状态信息，但是服务器依然会崩掉。
from flask import Flask,request,render_template
app = Flask(__name__)
# 捕获异常，如果出现异常，而且状态就是@app.errorhandler(404),
@app.errorhandler(404)
def error_404(arg):
print(arg)
return "404错误了"
@app.errorhandler(500)
def error(arg):
print(arg)
return "500错误了"
@app.route('/index')
def index():
print("我是真的视图")
return render_template("index.html")
if __name__ == '__main__':
# app.__call__
app.run()
5. template_global
from flask import Flask,request,render_template
app = Flask(__name__)
@app.template_global()
def plus(a1, a2):
return a1 + a2
@app.route('/index')
def index():
print("我是真的视图")
# 1.一般在模板语言中调用函数，需要穿函数地址
# return render_template("index.html",func=plus)
# 2.使用template_global不需要传别调用的函数地址且全局可调用plus函数
return render_template("index.html")
if __name__ == '__main__':
# app.__call__
app.run()
templates/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
{{ plus(1,2) }}
</body>
</html>
6. template_filter
from flask import Flask,request,render_template
app = Flask(__name__)
@app.template_filter()
def plus(a1, a2, a3):
print(a1,a2,a3) # 1 2 3
return a1 + a2 + a3
@app.route('/index')
def index():
print("我是真的视图")
return render_template("index.html")
if __name__ == '__main__':
# app.__call__
app.run()
templates/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
{{ 1|plus(2,3) }} <!--可以传任意多个参数-->
</body>
</html>
自定义中间件
from flask import Flask
app = Flask(__name__)
@app.route('/')
def index():
return 'Hello World!'
# 模拟中间件
class MiddleWare(object):
def __init__(self, old_wsgi_app):
self.old_wsgi_app = old_wsgi_app
def __call__(self, environ, start_response):
print('开始之前')
ret = self.old_wsgi_app(environ, start_response)
print('结束之后')
return ret
if __name__ == '__main__':
# 1我们发现当执行app.run方法的时候，最终执行run_simple，最后执行app(),也就是在执行app.__call__方法
# 2 在__call__里面，执行的是self.wsgi_app().那我们希望在执行他本身的wsgi之前做点事情。
# 3 所以我们先用MiddleWare类中__init__，保存之前的wsgi,然后我们又将app.wsgi转化成MiddleWared的对象。
# 4 那执行新的的app.wsgi_app，就是执行MiddleWare的__call__方法。
# 把原来的wsgi_app替换为自定义的，
app.wsgi_app = MiddleWare(app.wsgi_app)
# app.__call__
app.run()
请求流程
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
# 把request，和session放进去
ctx.push()
error = None
try:
try:
# 执行视图函数，请求扩展
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
# 返回响应
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
# 把request，和session删除
ctx.auto_pop(error)
def __call__(self, environ, start_response):
"""Shortcut for :attr:`wsgi_app`."""
return self.wsgi_app(environ, start_response)
多线程threading---Local
多个线程同时操作同一个全局变量，每次操作都会修改值，存在锁机制。导致每个线程取到的起始数据不相同，导致最后结果输出错误。
# 未使用local
from threading import Thread
import time
test = 10
def task(arg):
global test
test = arg
# 模拟IO操作，用于实现对全局变量的混乱操作
time.sleep(2)
print(test)
for i in range(10):
t = Thread(target=task, args=(i,))
t.start()
'''
打印结果：
10个9
'''
local类的简单使用
from threading import Thread
from threading import local
import time
from threading import get_ident # get_ident用于获取线程、进程ID号
# 初始化local对象
loc= local()
def task(arg):
# 对象.val = 1/2/3/4/5
# cxw['线程id']['value']=arg 内部做了处理
loc.value = arg
time.sleep(2)
print(loc.value)
for i in range(10):
t = Thread(target=task,args=(i,))
'''
3
1
0
2
5
4
7
6
9
8
'''
自定义字典实现Local的操作
from threading import get_ident,Thread
import time
storage = {}
def set(k,v):
ident = get_ident()
if ident in storage:
storage[ident][k] = v
else:
storage[ident] = {k:v}
def get(k):
ident = get_ident() # 根本不存在线程号，是生成一个唯一的整型数字识别号
return storage[ident][k]
def task(arg):
set('val',arg)
v = get('val')
print(v)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
基于面向对象自定义实现Local
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def set(self, k, v):
ident = get_ident() # 根本不存在线程号，是生成一个唯一的整型数字识别号
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def get(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.set('val',arg)
v = obj.get('val')
print(v)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
基于魔法方法实现Local
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def __setattr__(self, k, v):
ident = get_ident() # 根本不存在线程号，是生成一个唯一的整型数字识别号
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
基于魔法方法实现单独存储空间的Local
from threading import get_ident,Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident() # 根本不存在线程号，是生成一个唯一的整型数字识别号
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
实现Local并使其兼容线程和协程（参考Flask源码）
try:
from greenlet import getcurrent as get_ident
except Exception as e:
from threading import get_ident
from threading import Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident() # 根本不存在线程号，是生成一个唯一的整型数字识别号
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
偏函数
from functools import partial
def test(a,b,c,d):
return a+b+c+d
tes=partial(test,1,2,)
print(tes(3,4)) # 10
# tes=partial(test,1,2,c=3,d=4)
# print(tes()) # 10
print(tes) # functools.partial(<function test at 0x00000000003E1EA0>, 1, 2)
请求上下文
'''
设计模式中的代理模式： 将ctx = RequestContext(environ)放到的Local()中，在从全局的LocalProxy类中取ctx中的request
app.__call__
1 全局的变量
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
current_app = LocalProxy(_find_app)
request = LocalProxy(partial(_lookup_req_object, "request"))
session = LocalProxy(partial(_lookup_req_object, "session"))
g = LocalProxy(partial(_lookup_app_object, "g"))
2 flask请求上下文全部与源码：
def wsgi_app(self, environ, start_response):
#environ请求相关的所有数据,ctx就是ResquestContext的对象，包含request
ctx = self.request_context(environ)
error = None
try:
try:
#把request放进去了,Local()
ctx.push()
#执行视图函数，请求扩展
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
#返回响应
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
##把request,和session删除
ctx.auto_pop(error)
2.1 ctx = self.request_context(environ)，这个environ是请求相关的
self.request_context(environ)这个话的本质是执行return RequestContext(self, environ)
现在传给RequestContext的变量self是当前app,environ是请求相关的
2.1.1我们发现RequestContext是一个类，类加括号是就实例化得到对象，这个对象里面有request,session等：
app-->就当前的flask对象，environ请求相关的
def __init__(self, app, environ, request=None, session=None):
self.app = app
if request is None:
request = app.request_class(environ)
self.request = request
self.url_adapter = None
try:
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes = None
self.session = session
self.preserved = False
self._preserved_exc = None
self._after_request_functions = []
2.1.2 这样我们执行ctx = self.request_context(environ)，得到是RequestContext的对象这个对象里面有request,session等，
2.2 ctx.push()的源码：
执行的是RequestContext里面的push方法： _request_ctx_stack.push(self)，当前的self是ctx
我们发现_request_ctx_stack就是LocalStack() ，_request_ctx_stack.push(self)就是执行LocalStack的push,并且把ctx传过来了
2.2.1 _request_ctx_stack.push(self)的源码：
#obj就是ctx,self._local是Local对象，用来区分不同线程，协程的数据
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
# self._local=>stack-->storage['线程id']['stack']=[]
self._local.stack = rv = []
# self._local=>stack-->storage['线程id']['stack']=[ctx,]
rv.append(obj)
return rv
3 request是怎样取到我放到 Local对象里面的ctx在里面的request
request = LocalProxy(partial(_lookup_req_object, "request"))
当我们request.form时候就会找LocalProxy(partial(_lookup_req_object, "request"))要form属性
3.1当要request属性时候，就会执行LocalProxy的getattr,源码：
#name=我们要的属性，比如request.form的form,name=form
def __getattr__(self, name):
if name == "__members__":
return dir(self._get_current_object())
#name-->form,self._get_current_object()===>ctx.request,form
return getattr(self._get_current_object(), name)
最后我们发现self._get_current_object()是我们在实例化LocalProxy(partial(_lookup_req_object, "request"))
传过来的偏函数执行结果。
3.1.1:self._get_current_object()要执行的代码是：
def _get_current_object(self):
if not hasattr(self.__local, "__release_local__"):
return self.__local()
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
3.1.1.1 self._get_current_object()这个里面的self.__local()：
我们发先这个self.__local是在实例化的时候传过来：
def __init__(self, local, name=None):
object.__setattr__(self, "_LocalProxy__local", local)
我们发现传过来的是partial(_lookup_req_object, "request")偏函数：
偏函数源码：
#name=request
def _lookup_req_object(name):
top = _request_ctx_stack.top
#top就是我们前面放进去ctx
if top is None:
raise RuntimeError(_request_ctx_err_msg)
#getattr(top, name)的意思就是到ctx里面找request
return getattr(top, name)
3.1.1.1.1 top = _request_ctx_stack.top:我们在前面已经知道_request_ctx_stack是 LocalStack()的对象：
源码：
@property
def top(self):
try:
#返回ctx
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
'''
蓝图
使用蓝图对flask应用进行目录划分
单应用APP
目录结构：
pro_flask
|——pro_flask
|——statics
|——css
|——img
|——js
|——templates
|——login.html
|——views
|————account.py
|————blog.py
|————user.py
|——__init__.py
|——run.py
run.py
from pro_flask import app
if __name__ == '__main__':
app.run()
init.py
from flask import Flask
app = Flask(__name__,template_folder='templates',static_folder='statics',static_url_path='/static')
# views中所有视图函数公用的请求扩展
@app.before_request
def before_req():
print("我是app里面的before_request")
from .views.account import account
from .views.blog import blog
from .views.user import user
# 注册蓝图
app.register_blueprint(account)
app.register_blueprint(blog)
app.register_blueprint(user)
views/account.py
from flask import Blueprint
from flask import render_template
from flask import request
account = Blueprint('acc', __name__)
@account.route('/login.html', methods=['GET', "POST"])
def login():
return render_template('login.html')
templates/login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>用户登录</h1>
<form method="POST">
<input type="text" name="user"/>
<input type="submit" value="提交"/>
</form>
<img src="/static/img/code.png">
<img src="{{ url_for('static',filename='code.png') }}">
</body>
</html>
views/blog.py
from flask import Blueprint,url_for,redirect
blog = Blueprint('blog', __name__)
@blog.before_request
def before_req():
print("我是blog 的请求扩展")
@blog.route("/index")
def index(): # 反向解析以Blueprint实例化传的别名为准。
re_url = url_for("acc.login")
print(url_for("acc.login"))
# return redirect(re_url)
return "ok"
user.py
from flask import Blueprint
user = Blueprint('user', __name__)
多应用app
目录结构：
pro_flask
|__ pro_flask
| |__ admin
| | |__ static
| | |__ templates
| | |__ __init__.py
| | |__ views.py
| |__ web
| | |__static
| | |__ templates
| | |__ __init__.py
| | |__ views.py
| |__ __init__.py
|__ run.py
run.py
from pro_flask import app
if __name__ == '__main__':
app.run()
pro_flask/init.py
from flask import Flask
from .admin import admin
from .web import web
app = Flask(__name__)
app.debug = True
app.register_blueprint(admin, url_prefix='/admin')
app.register_blueprint(web)
pro_flask/admin/init.py
from flask import Blueprint
admin = Blueprint(
'admin',
__name__,
template_folder='templates',
static_folder='static'
)
# 必须放在下面，不然在项目下pro_flask下的__init__.py文件
# 下导入蓝图对象时找不到，报错。
from . import views
pro_flask/admin/views.py
from . import admin
@admin.route('/index')
def index():
return 'Admin.Index'
pro_flask/web/init.py
from flask import Blueprint
web = Blueprint(
'web',
__name__,
template_folder='templates',
static_folder='static'
)
# 必须放在下面，不然在项目下pro_flask下的__init__.py文件
# 下导入蓝图对象时找不到，报错。
from . import views
pro_flask/web/views.py
from . import web
@web.route('/index')
def index():
return 'Web.Index'
g对象
g对象专门用来存储用户信息的g对象，在一次请求中的所有代码的地方，都是可以使用的。同一用户再次请求别别的接口，这个g对象存储的用户信息就没了。与session还是有区别的。
session对象是可以跨request的，只要session还未失效，不同的request的请求会获取到同一个session，但是g对象不是，g对象不需要管过期时间，请求一次就g对象就改变了一次，或者重新赋值了一次
from flask import Flask,request,g,redirect
app=Flask(__name__)
def set_g():
g.name='zhang'
@app.route("/")
def index():
set_g()
print(g.name) # 可以取多次，flash(闪现)只能取一次
print(g.name)
print(g.name)
return 'name:%s'%g.name
# return redirect("/index") # 通过跳转到login视图函数也不能取到g.name 会报错
@app.route("/index")
def login():
print(g.name)
return "2"
if __name__ == '__main__':
app.run()
信号
作用：可以在flask请求过程中定制一些用户行为
使用信号首先要先安装blinker模块：pip install blinker
from flask import Flask,signals,render_template
app = Flask(__name__)
# 往信号中注册函数
#1给信号绑定要执行的函数
#无需管调用，因为flask,已经给我们设置调用点
def func(*args,**kwargs):
print('触发型号',args,kwargs)
#与该信号进行绑定
signals.request_started.connect(func)
# signals.request_started.send
# 触发信号： signals.request_started.send()
@app.before_first_request
def before_first(*args,**kwargs):
print("befor_first_request")
@app.before_request
def before_req(*args,**kwargs):
print("befor_request")
@app.route('/',methods=['GET',"POST"])
def index():
print('index视图')
return "index视图"
if __name__ == '__main__':
# app.wsgi_app
app.run()
其他信号：
request_started = _signals.signal('request-started') # 请求到来前执行
request_finished = _signals.signal('request-finished') # 请求结束后执行
before_render_template = _signals.signal('before-render-template') # 模板渲染前执行
template_rendered = _signals.signal('template-rendered') # 模板渲染后执行
got_request_exception = _signals.signal('got-request-exception') # 请求执行出现异常时执行
request_tearing_down = _signals.signal('request-tearing-down') # 请求执行完毕后自动执行（无论成功与否）
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 应用上下文执行完毕后自动执行（无论成功与否）
appcontext_pushed = _signals.signal('appcontext-pushed') # 应用上下文push时执行
appcontext_popped = _signals.signal('appcontext-popped') # 应用上下文pop时执行
message_flashed = _signals.signal('message-flashed') # 调用flask在其中添加数据时，自动触发
整个流程当中的信号触发点：
a. before_first_request
b. 触发 request_started 信号
c. before_request
d. 模板渲染
渲染前的信号 before_render_template.send(app, template=template, context=context)
rv = template.render(context) # 模板渲染
渲染后的信号 template_rendered.send(app, template=template, context=context)
e. after_request
f. session.save_session()
g. 触发 request_finished信号
如果上述过程出错：
触发错误处理信号 got_request_exception.send(self, exception=e)
h. 触发信号 request_tearing_down
注意：可以格局实际需要，自定义信号，具体参见flask官方文档；
from flask.signals import _signals
aa = _signals.signal('aa') # 自定义的信号
flask_session
使用：先安装flask_session pip install flask_session
作用：将默认 保存的签名cookie中的值，保存到redis、memcached/file/mongodb/SQLAlchemy
保存session方式一：
from flask import Flask,session
from flask_session import RedisSessionInterface
import redis
app = Flask(__name__)
conn=redis.Redis(host='127.0.0.1',port=6379, db=6)
# 参数use_signer是否对key签名，设置成False就不需要app.secret_key,不然就像使用session一样要设置。
# key_prefix 保存的随机字符串的前缀
# 关闭浏览器session值是否失效
app.session_interface=RedisSessionInterface(conn,key_prefix='abc', use_signer=False, permanent=True)
@app.route('/')
def hello_world():
session['name']='zhang'
return 'Hello World!'
@app.route('/index')
def index():
print(session['name'])
print(session['name'])
return 'index'
if __name__ == '__main__':
app.run()
保存session方式二：
from flask import Flask,session
import redis
from flask_session import Session
app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] =redis.Redis(host='127.0.0.1',port='6379')
app.config['SESSION_KEY_PREFIX']="xiaowang"
app.config['SESSION_USE_SIGNER'] = False
app.config['SESSION_PERMANENT'] =True
Session(app)
@app.route('/')
def hello_world():
session['password']='123'
return '密文!'
@app.route("/index")
def index():
print(session['password'])
return "ok"
if __name__ == '__main__':
app.run()
使用flask_script自定义终端命令
首先，安装插件：pip install flask_script
from flask_script import Manager
from flask import Flask
app = Flask(__name__)
manager = Manager(app)
@app.route('/')
def index():
return 'ok'
@manager.command
def my_command(a,b) # 接收命令行命令传过来的参数 zhang 18
print(a,b)
if __name__ == '__main__':
manager.run()
'''
在终端启动脚本：python flask_script_shell.py runserver
就像使用django一样，有一个启动的入口文件manage.py,这里flask_script_shell.py就是flask启动的入口文件
'''
自定制命令
函数要接收多少个参数，就要写多少个装饰器，其中-n -u 是简写，--name --url是全写，dest是关键字参数。
from flask_script import Manager
from flask import Flask
app = Flask(__name__)
manager = Manager(app)
@manager.option('-n', '--name', dest='name')
@manager.option('-u', '--url', dest='url')
def cmd(name, url):
"""
自定义命令（-n也可以写成--name）
执行： python flask_script_shell.py cmd -n zhang -u http://www.baidu.com
:param name:
:param url:
:return:
"""
print(name, url) # zhang http://www.baidu.com
#有什么用？
#把excel的数据导入数据库，定制个命令，去执行
wtforms组件
类似于django的form组件，可以对请求的数据包数据进行校验，但是wtforms并不是flask框架内嵌校验的模块，所以我们在使用时要先安装：pip install wtforms
注册校验：
register.py
from flask import Flask, render_template, request, redirect
from wtforms import Form
from wtforms.fields import core
from wtforms.fields import html5
from wtforms.fields import simple
from wtforms import validators
from wtforms import widgets
app = Flask(__name__, template_folder='templates')
app.debug = True
class RegisterForm(Form):
def validate_pwd_confirm(self, field): # 全局钩子校验密码是否一致
"""
自定义pwd_confirm字段规则，例：与pwd字段是否一致
:param field:
:return:
"""
# 最开始初始化时，self.data中已经有所有的值
print(field.data)
if field.data !="zhang":
#raise validators.ValidationError("zhang") # 继续后续验证
raise validators.StopValidation("zhang") # 不再继续后续验证
# if field.data != self.data['pwd']:
# raise validators.ValidationError("密码不一致") # 继续后续验证
#raise validators.StopValidation("密码不一致") # 不再继续后续验证
name = simple.StringField(
label='用户名',
validators=[
validators.DataRequired()
],
widget=widgets.TextInput(),
render_kw={'class': 'form-control'},
default='zhang'
)
pwd = simple.PasswordField(
label='密码',
validators=[
validators.DataRequired(message='密码不能为空.')
],
widget=widgets.PasswordInput(),
render_kw={'class': 'form-control'}
)
pwd_confirm = simple.PasswordField(
label='重复密码',
validators=[
validate_pwd_confirm,
validators.DataRequired(message='重复密码不能为空.'),
#validators.EqualTo('pwd', message="两次密码输入不一致")
],
widget=widgets.PasswordInput(),
render_kw={'class': 'form-control'}
)
email = html5.EmailField(
label='邮箱',
validators=[
validators.DataRequired(message='邮箱不能为空.'),
validators.Email(message='邮箱格式错误')
],
widget=widgets.TextInput(input_type='email'),
render_kw={'class': 'form-control'}
)
gender = core.RadioField(
label='性别',
choices=(
(1, '男'),
(2, '女'),
),
coerce=int # “1” “2”
)
city = core.SelectField(
label='城市',
choices=(
('bj', '北京'),
('sh', '上海'),
)
)
hobby = core.SelectMultipleField(
label='爱好',
choices=(
(1, '篮球'),
(2, '足球'),
),
coerce=int
)
favor = core.SelectMultipleField(
label='喜好',
choices=(
(1, '篮球'),
(2, '足球'),
),
widget=widgets.ListWidget(prefix_label=False),
option_widget=widgets.CheckboxInput(),
coerce=int,
default=[1, 2]
)
def __init__(self, *args, **kwargs):
super(RegisterForm, self).__init__(*args, **kwargs)
self.favor.choices = ((1, '篮球'), (2, '足球'), (3, '羽毛球'))
self.favor.data=[1,]
@app.route('/register', methods=['GET', 'POST'])
def register():
if request.method == 'GET':
form = RegisterForm(data={'gender': 2,'hobby':[1,]}) # initial
return render_template('register.html', form=form)
else:
form = RegisterForm(formdata=request.form)
if form.validate():
print('用户提交数据通过格式验证，提交的值为：', form.data)
else:
print(form.errors)
return render_template('register.html', form=form)
if __name__ == '__main__':
app.run()
templates/register.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>用户注册</h1>
<form method="post" novalidate style="padding:0 50px">
{% for field in form %}
<p>{{field.label}}: {{field}} {{field.errors[0] }}</p>
{% endfor %}
<input type="submit" value="提交">
</form>
</body>
</html>
登录校验:
login.py
from flask import Flask, render_template, request, redirect
from wtforms import Form
from wtforms.fields import simple
from wtforms import validators
from wtforms import widgets
app = Flask(__name__)
app.debug = True
class LoginForm(Form):
# 字段（内部包含正则表达式）
name = simple.StringField(
label='用户名',
validators=[
validators.DataRequired(message='用户名不能为空.'),
validators.Length(min=2, max=6, message='用户名长度必须大于%(min)d且小于%(max)d')
],
widget=widgets.TextInput(), # 页面上显示的插件
render_kw={'class': 'form-control'} # 控制输入框的显示格式
)
# 字段（内部包含正则表达式）
pwd = simple.PasswordField(
label='密码',
validators=[
validators.DataRequired(message='密码不能为空.'),
validators.Length(min=8, message='密码长度必须大于%(min)d'),
# validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*d)(?=.*[$@$!%*?&])[A-Za-zd$@$!%*?&]{8,}",
# message='密码至少8个字符，至少1个大写字母，1个小写字母，1个数字和1个特殊字符')
],
widget=widgets.PasswordInput(),
render_kw={'class': 'form-control'}
)
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'GET':
form = LoginForm()
return render_template('login.html', form=form)
else: # post请求
form = LoginForm(formdata=request.form)
if form.validate():
print('用户提交数据通过格式验证，提交的值为：', form.data)
else:
print(form.errors)
return render_template('login.html', form=form)
if __name__ == '__main__':
app.run()
templates/login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>登录</h1>
<form method="post">
<p>{{form.name.label}} {{form.name}} {{form.name.errors[0] }}</p>
<p>{{form.pwd.label}} {{form.pwd}} {{form.pwd.errors[0] }}</p>
<input type="submit" value="提交">
</form>
</body>
</html>
sqlalchemy 的使用
Python中最有名的ORM架构就是SQLAlchemy。
ORM，对象-关系映射（Object/Relation Mapping，简称ORM），是随着面向对象的软件开发方法发展而产生的。面向对象的开发方法是当今企业级应用开发环境中的主流开发方法，关系数据库是企业级应用环境中永久存放数据的主流数据存储系统。对象和关系数据是业务实体的两种表现形式，业务实体在内存中表现为对象，在数据库中表现为关系数据。内存中的对象之间存在关联和继承关系，而在数据库中，关系数据无法直接表达多对多关联和继承关系。因此，对象-关系映射(ORM)系统一般以中间件的形式存在，主要实现程序对象到关系数据库数据的映射。
建表和删表
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship
Base = declarative_base()
# 定义的类（表），必须继承Base类，才能创建表。 创建表之前先自己手动创建数据库。
class Users(Base):
__tablename__ = 'users' # 数据库表名称
id = Column(Integer, primary_key=True) # id 主键 默认主键自增
name = Column(String(32), index=True, nullable=False) # name列，索引，不可为空
# email = Column(String(32), unique=True)
age = Column(Integer,default=0)
#datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
ctime = Column(DateTime, default=datetime.datetime.now)
# extra = Column(Text, nullable=True)
# __table_args__ = (
# UniqueConstraint('id', 'name', name='uix_id_name'), #联合唯一
# Index('ix_id_name', 'name', 'email'), #索引
# )
def __repr__(self):
return self.name
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
class Person(Base):
__tablename__ = 'person'
nid = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
# hobby指的是tablename而不是类名，
hobby_id = Column(Integer, ForeignKey("hobby.id"))
# 跟数据库无关，不会新增字段，只用于快速链表操作
# 类名，backref用于反向查询,uselist=False
hobby = relationship('Hobby', backref='pers')
def __repr__(self): # 输出打印格式比__str__更底层， 打印对象时触发该方法，还有一个__str__在这里不能用，但django可以
return self.name
class Boy2Girl(Base):
__tablename__ = 'boy2girl'
id = Column(Integer, primary_key=True, autoincrement=True)
girl_id = Column(Integer, ForeignKey('girl.id'))
boy_id = Column(Integer, ForeignKey('boy.id'))
class Girl(Base):
__tablename__ = 'girl'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
class Boy(Base):
__tablename__ = 'boy'
id = Column(Integer, primary_key=True, autoincrement=True)
hostname = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
girl = relationship('Girl', secondary='boy2girl', backref='boys')
# 创建表，表一旦被创建，就不能在表中添加额外的字段执行添加表字段操作。
# 只能手动使用mysql客户端或者Navicat工具添加字段，添加的字段的规则必须
# 和我们定义的类中定义的一致，不然数据无法入库
def init_db():
"""
根据类创建数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.create_all(engine)
# 删除表
def drop_db():
"""
根据类删除数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
# drop_db() # 删除表
init_db() # 创建表
表添加数据的简单操作
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#"mysql+pymysql://root@127.0.0.1:3306/aaa"
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/flask", max_overflow=0, pool_size=5)
Connection = sessionmaker(bind=engine)
# 每次执行数据库操作时，都需要创建一个Connection
con = Connection()
# ############# 执行ORM操作 #############
obj1 = Users(name="zhang",age=20)
con.add(obj1)
# 提交事务
con.commit()
# 关闭session，其实是将连接放回连接池
con.close()
以上增加数据的操作，假如如果是多个客户端连接数据库，开启多个线程，一个线程A进行增操作，这时由于网络IO，另一个线程B也进行了增操作，并提交；这时问题就来了，这个线程B就把线程A的数据也commit掉了，sqlalchemy引擎并不知道是哪个线程提交的数据，数据的操作就不安全啦！这个时候我们就想到请求上下文中的Local（）对象，来保证线程的安全。实际上，这个问题sqlachemy已经帮我们解决了，我们只要熟悉它是如何实现的就行了。它就是scoped_session。
使用scoped_session保证数据操作线程的安全
在使用sqlalchemy时，为了线程的安全，一定要使用scoped_session
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Users
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy",
max_overflow=0,
pool_size=5)
Session = sessionmaker(bind=engine)
"""
# 线程安全，基于本地线程实现每个线程用同一个session
# 特殊的：scoped_session中有原来方法的Session中的以下方法：
public_methods = (
'__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested',
'close', 'commit', 'connection', 'delete', 'execute', 'expire',
'expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind',
'is_modified', 'bulk_save_objects', 'bulk_insert_mappings',
'bulk_update_mappings',
'merge', 'query', 'refresh', 'rollback',
'scalar'
)
"""
#scoped_session类并没有继承Session,但是却又它的所有方法
session = scoped_session(Session)
# ############# 执行ORM操作 #############
obj1 = Users(name="zhang") # age 不传默认值是0
session.add(obj1)
# 提交事务
session.commit()
# 关闭session
session.close()
单表操作
群增
import time
import threading
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from models import Users,Hobby
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy",
max_overflow=0,
pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
'''# 1.单增操作'''
user_obj = Users(name='李四', age=88)
session.add(user_obj)
'''# 2.群增操作'''
user_obj = session.add_all([
Users(name='王五',age=35),
# Users(name='翠花',age=17),
# 如果还有其他表的数据也可以一并传进去。
# Teacher(name='李晓霞'telephone='1234567',)
])
print(user_obj) # 新增成功返回None，失败直接抛异常。
'''# 3.删除操作 query(Users)指定要操作的表 filter(Users.id >= 4):查询条件'''
user_obj = session.query(Users).filter(Users.id >= 12).delete()
print(user_obj) # 删除成功返回1，否则0
'''4.更新操作 要传一个字典'''
session.query(Users).filter(Users.id ==1).update({'name':'张', 'age':18})
# 类似django的F查询 如果是追加的字符串，需要添加一个属性synchronize_session=False，不然报错。
obj = session.query(Users).filter(Users.id == 2).update({Users.name: Users.name + '999'},synchronize_session=False)
print(obj) # 1 代表更新成功
# 如果拼接的参数是整型，需要添加一个属性synchronize_session=evaluate，不然报错。
obj = session.query(Users).filter(Users.id == 2).update({Users.age: Users.age +8},synchronize_session='evaluate')
print(obj) # 1 代表修改成功 0 代表修改失败
'''5.查询'''
# 可以同时查询多个表 不建议，因为最终查询的结果
# 是列表套元组[(user_obj,hobby_obj),(user_obj,hobby_obj),],
# 引起结果错乱
user_list = session.query(Users, Hobby).all() # 可以同时查询多个表
print(user_list) # [(user_obj,hobby_obj),(user_obj,hobby_obj),],
user_list = session.query(Users).all() # 列表套字典
print(user_list) # [<models.Users object at 0x000000000B31FE10>, <models.Users object at 0x000000000B31FE80>, <models.Users object at 0x000000000B31FEF0>]
for user_obj in user_list:
print(user_obj.name,user_obj.age)
user_list = session.query(Users)
print(user_list) # 不加all()返回原生的sql语句
# 只取age列，把name重命名为xx
user_obj = session.query(Users.name.label('username'),Users.age).filter(Users.id == 3).first()
print(user_obj) # 元组 ('李四', 88)
print('username:',user_obj.username, 'age:',user_obj.age) # username: 李四 age: 88
# filter 传的是表达式，filter_by传的是参数
user_list1 = session.query(Users).filter(Users.name == '张').all()
print(user_list1)
# all()返回的是列表套对象，first()直接返回对象
user_list2 = session.query(Users).filter_by(name='张').first()
print(user_list2)
# :value 和:name 相当于占位符，用params传参数
user_list1 = session.query(Users).filter(text('id<:value and name=:name')).params(value=2,name='张').order_by(Users.id).all()
print(user_list1) # [<models.Users object at 0x000000000B3EE390>]
# 自定义查询SQL
user_list2 = session.query(Users).from_statement(text('SELECT * FROM users where name=:name')).params(name='张').all()
print(user_list2) # [<models.Users object at 0x000000000B32AE48>]
'''6.多条件表达式，之间是and连接'''
user_list = session.query(Users).filter(Users.id > 1, Users.name == '李四').all()
print(user_list) # [<models.Users object at 0x000000000B3620B8>]
'''7.between的使用'''
user_list = session.query(Users).filter(Users.id.between(1, 3), Users.age <= 18).all()
print(user_list) # [<models.Users object at 0x000000000B360198>, <models.Users object at 0x000000000B360208>]
'''8.in 的使用'''
user_list = session.query(Users).filter(Users.id.in_([1,3,4])).all()
print(user_list) # [<models.Users object at 0x000000000B322FD0>, <models.Users object at 0x000000000B360080>]
'''9. ~ 非，除..外'''
user_list = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
print(user_list) # [<models.Users object at 0x000000000B3600F0>]
'''10. 子查询'''
user_list = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='张'))).all()
print(user_list) # [<models.Users object at 0x000000000B3630F0>]
'''11. and 和or的使用'''
from sqlalchemy import and_, or_
# #or_包裹的都是or条件，and_包裹的都是and条件
user_list = session.query(Users).filter(and_(Users.id > 1, Users.name == 'wang999')).all()
print(user_list) # [wang999] __repr__
user_list = session.query(Users).filter(or_(Users.id < 3, Users.name == '张')).all()
print(user_list) # [张, wang999]
user_list = session.query(Users).filter(
# or_(
# Users.id < 2,
# and_(Users.name == '张', Users.id > 3),
# Users.age != ""
# )).all()
# print(user_list) # [张, wang999, 李四]
'''12. 模糊匹配like，以e开头，不以e开头 '''
user_list = session.query(Users).filter(Users.name.like('w%')).all()
print(user_list) # [wang999]
user_list = session.query(Users).filter(~Users.name.like('w%')).all()
print(user_list) # [wang999] # [张, 李四]
'''13. 限制，用于分页，区间 顾头不顾尾'''
user_list = session.query(Users)[1:2]
print(user_list) # [wang999]
'''14. 排序order_by，根据name降序排列（从大到小）'''
user_list = session.query(Users).order_by(Users.age.desc()).all()
print(user_list) # [李四, 张, wang999]
# 多重拍讯条件， 第一个条件重复后，再按第二个条件升序排
user_list = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
print(user_list)
'''15. 分组group_by'''
from sqlalchemy.sql import func
user_list = session.query(Users).group_by(Users.age).all()
print(user_list) # [wang999, 张, 李四]
#分组之后取最大id，id之和，最小id
user_list = session.query(
func.max(Users.age),
func.sum(Users.age),
func.min(Users.age)).group_by(Users.name).all()
print(user_list)
'''16. 筛选having'''
user_list = session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >=2).all()
print(user_list) # [(2, Decimal('2'), 2), (3, Decimal('3'), 3)]
'''17. 连表（默认用foreign key关联）'''
user_list = session.query(Users, Hobby).filter(Users.id == Hobby.id).all()
print(user_list)
# [(张, <models.Hobby object at 0x000000000B350908>),
# (wang999, <models.Hobby object at 0x000000000B3509E8>),
# (李四, <models.Hobby object at 0x000000000B350AC8>)]
'''18. join表，默认是inner join'''
user_list = session.query(Users).join(Hobby).all()
print(user_list)
# 左连接， isouter=True 外连，表示Users left joinHobby，没有右连接，反过来即可
user_list = session.query(Users).join(Hobby,Users.id==Hobby.id, isouter=True).all()
print(user_list) # [张, wang999, 李四]
# 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
user_list = session.query(Users).join(Hobby,Users.id==Hobby.id, isouter=True).all()
print(user_list) # [张, wang999, 李四]
# # 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
# #union和union all的区别？
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Hobby.caption).filter(Hobby.id < 2)
user_list = q1.union(q2).all()
print(user_list) # [('李四',), ('羽毛球',)]
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Hobby.caption).filter(Hobby.id < 2)
user_list = q1.union_all(q2).all()
print(user_list) # [('李四',), ('羽毛球',)]
session.commit()
session.close()
一对多关系查询
建表：
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship
Base = declarative_base()
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
class Person(Base):
__tablename__ = 'person'
nid = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
# hobby指的是tablename而不是类名，
hobby_id = Column(Integer, ForeignKey("hobby.id"))
# 跟数据库无关，不会新增字段，只用于快速链表操作
# 类名，backref用于反向查询,uselist=False
hobby = relationship('Hobby', backref='pers')
def __repr__(self):
return self.name
def init_db():
"""
根据类创建数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.create_all(engine)
# 删除表
def drop_db():
"""
根据类删除数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
# drop_db() # 删除表
init_db() # 创建表
插入数据并查询：两张表建立在一个爱好被多人喜欢的一对多关系。一对多放在多的一方，即Person那张表
import time
import threading
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from models import Users, Hobby, Person
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy?charset=utf8",
max_overflow=0,
pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
'''
# 第一种添加方式： 一次给多表添加多条记录
session.add_all([
Hobby(caption='乒乓球'),
Hobby(caption='羽毛球'),
Person(name='张三', hobby_id=1),
Person(name='李四', hobby_id=2),
])
'''
'''
# 第二种添加方式：正向插入
person = Person(name='张九', hobby=Hobby(caption='姑娘'))
session.add(person)
'''
'''
# 第三种添加方式：反向插入
hb = Hobby(caption='人妖')
hb.pers = [Person(name='文飞'), Person(name='博雅')] # 相当于django表设计的related_name
session.add(hb)
session.commit()
session.close()
'''
# 使用relationship正向查询
# user_obj = session.query(Person).first()
# print(user_obj.name) # 张三
# print(user_obj.hobby.caption) # 乒乓球
# 使用relationship反向查询
# hobby_obj = session.query(Hobby).first()
# print(hobby_obj.caption) # 乒乓球
# print(hobby_obj.pers) # [张三]
#方式一，自己链表
# person_list=session.query(Person).join(Hobby,isouter=True)
# print(person_list)
'''
SELECT person.nid AS person_nid, person.name AS person_name, person.hobby_id AS person_hobby_id
FROM person LEFT OUTER JOIN hobby ON hobby.id = person.hobby_id
SELECT person.nid AS person_nid, person.name AS person_name, person.hobby_id AS person_hobby_id, hobby.id AS hobby_id, hobby.caption AS hobby_caption
FROM person LEFT OUTER JOIN hobby ON hobby.id = person.hobby_id
'''
# person_list=session.query(Person,Hobby).join(Hobby,isouter=True)
# # print(person_list)
# for row in person_list:
# print(row[0].name,row[1].caption)
'''
张三 乒乓球
李四 羽毛球
张九 姑娘
文飞 人妖
博雅 人妖
'''
#方式二：通过relationship
# person_list=session.query(Person).all()
# for row in person_list:
# print(row.name,row.hobby.caption)
#查询喜欢姑娘的所有人
# obj=session.query(Hobby).filter(Hobby.caption=='人妖').first()
# user_list=obj.pers
# for person in user_list:
# print(person.nid,person.name,)
# '''
# 4 文飞
# 5 博雅
# '''
# session.close()
# 如果没有建立外键 join 连接（这里是左连接，如果表的数据量非常大，连接表非常耗时，不推荐）
ret = session.query(Person).join(Hobby,Person.hobby_id==Hobby.id, isouter=True).filter(Person.nid==1).all()
print(ret) # [张三]
多对多关系查询
建表：
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship
Base = declarative_base()
'''
创建表，表一旦被创建，就不能在表中添加额外的字段类执行添加表字段操作。
必须手动使用mysql客户端或者Navicat工具添加字段，添加的字段的规则必须
和我们定义的类中定义的一致，不然数据无法入库
'''
class Boy2Girl(Base):
__tablename__ = 'boy2girl'
id = Column(Integer, primary_key=True, autoincrement=True)
girl_id = Column(Integer, ForeignKey('girl.id'))
boy_id = Column(Integer, ForeignKey('boy.id'))
class Girl(Base):
__tablename__ = 'girl'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
class Boy(Base):
__tablename__ = 'boy'
id = Column(Integer, primary_key=True, autoincrement=True)
hostname = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以,
# secondary 多对多关系当中指向的第三张表，backref为反向查询的字段
girl = relationship('Girl', secondary='boy2girl', backref='boys')
def init_db():
"""
根据类创建数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.create_all(engine)
# 删除表
def drop_db():
"""
根据类删除数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
# drop_db() # 删除表
init_db() # 创建表
多对多插入数据并查询：
import time
import threading
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from models import Users, Hobby, Person, Girl, Boy2Girl, Boy
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/flask?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
# session.add_all([
# Girl(name='c1.com'),
# Girl(name='c2.com'),
# Boy(hostname='A组'),
# Boy(hostname='B组'),
# ])
# session.commit()
# s2g = Boy2Girl(girl_id=1,boy_id=1)
# session.add(s2g)
# session.commit()
#
#
# gp = Girl(name='C组')
# gp.boys = [Boy(hostname='c3.com'),Boy(hostname='c4.com')]
# session.add(gp)
# session.commit()
#
#
# ser = Boy(hostname='c6.com')
# ser.girl = [Girl(name='F组'),Girl(name='G组')]
# session.add(ser)
# session.commit()
# 使用relationship正向查询
# v = session.query(Boy).first()
# print(v.hostname)
# print(v.girl)
# 使用relationship反向查询
# v = session.query(Girl).first()
# print(v.name)
# print(v.boys)
#
#
#
# session.close()
flask_sqlalchemy
flask本身并不支持使用像django一样的命令来生成迁移文件（像django的命令，python manage.py migrat....）， 和模型类生成数据库表，我们有第三方模块可以安装，达到和django一样使用命令的效果，执行迁移文件、生成数据库表、启动flask项目。
封装成和django一样的文件目录格式：
1.项目目录：
pro_flask_sansa__ # 项目根目录
|__ sansa
| |__ views
| | |__account.py
| |__ __init__.py
| |__models.py
|__manage.py
|__settings.py
sansa/views/account.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Blueprint
from .. import db
from .. import models
account = Blueprint('account', __name__)
@account.route('/register')
def register():
# 在 sqlalchemy进行初始化，生成db对象时就将session封装到db对象中了。
# self.session = self.create_scoped_session(session_options)，调用create_scoped_session方法，
# 这样操作数据库也是线程安全的，
user_obj1 = db.session.add(models.Users(username='wangwu', email='wangwu@qq.com'))
print(user_obj1) # 增加成功返回None
user_list = db.session.query(models.Users).all()
print(user_list) # [<User 'zhang'>, <User 'zhangsan'>, <User 'wangwu'>]
# 添加示例
"""
db.session.add(models.Users(username='王大妈', pwd='123', gender=1))
db.session.commit()
obj = db.session.query(models.Users).filter(models.Users.id == 1).first()
print(obj)
PS: db.session和db.create_session
"""
# db.session.add(models.Users(username='王大妈', email='da@.com'))
# db.session.add(models.Users(username='王二妈', email='er@.com'))
# db.session.add(models.Users(username='王三妈',email=san@qq.com'))
# db.session.commit()
# user_list = db.session.query(models.Users).all()
db.session.close()
# for item in user_list:
# print(item.username)
return 'register'
sansa/init.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
from .models import *
from .views import account
def create_app():
app = Flask(__name__)
app.config.from_object('settings.DevelopmentConfig')
# 将db注册到app中
db.init_app(app)
# 注册蓝图
app.register_blueprint(account.account)
return app
sansa/models.py
'''
只需要导入SQLAlchemy的实例化对象db,就可以解
决表及表字段设计的所有问题，不用再像之前直接使用SQLAlchemy一样，需
要导入各种字段类型，表关系规则之类的所有东西了，而且django表设计有些像了
可以对设计的表字段进行注释，就可以从数据库中直接删除该表的字段信息。
'''
from . import db
# self.Model = self.make_declarative_base(model_class, metadata)
# 跟从sqlalchemy导入declarative_base有点像，然后实例
# 化Base(Base = declarative_base())定义的模型表继承当前Base类，
# 检查make_declarative_base函数发现内部使用的还是declarative_base
class Users(db.Model):
"""
用户表
"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=True)
ids = db.Column(db.Integer) # 删除表中该字段信息
def __repr__(self):
return '<User %r>' % self.username
manage.py
from sansa import create_app
from flask_script import Manager
from flask_migrate import Migrate,MigrateCommand
from sansa import db
app = create_app()
manager=Manager(app)
#为了实现迁移
Migrate(app,db)
#现在把命令注册进来
manager.add_command('db1', MigrateCommand)
if __name__ == '__main__':
app.run() # 开发测试环境，右键直接运行即可，方便快捷
# manager.run() # 项目上线换成这种方式（使用命令行启动：python manage.py runserver）
settings.py
class BaseConfig(object): # 公共配置
# SESSION_TYPE = 'redis' # session类型为redis
# SESSION_KEY_PREFIX = 'session:' # 保存到session中的值的前缀
# SESSION_PERMANENT = True # 如果设置为False，则关闭浏览器session就失效。
# SESSION_USE_SIGNER = False # 是否对发送到浏览器上 session:cookie值进行加密
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:123@127.0.0.1:3306/flask_sqlalchemy_app?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False
class ProductionConfig(BaseConfig): # 生产环境配置
pass
class DevelopmentConfig(BaseConfig): # 开发环境配置
pass
class TestingConfig(BaseConfig): # 测试环境的配置
pass
2.安装：
pip install flask_sqlalchemy,
pip install flask-migrate 生成迁移文件
3.在以上表数据准备好的情况下，我们执行迁移文件：
跟django有点区别，首先执行：
python3 manage.py db1 init # 生成迁移文件 相当于makemigrations
然后执行，python manage.py db1 migrate
然后第三步执行：python manage.py db1 upgrade # 将模型表迁移到数据库中。相当于django中的python manage.py migrate
如果想要给表添加或删除字段，再或者添加字段规则只需要添加相应的字段或注释该字段，添加字段类的规则或去除规则，然后顺序执行一下两个命令就可以了。
python manage.py db1 migrate
python manage.py db1 upgrade
启动flask项目命令和django统一了：python manage.py runserver
服务端访问:http://127.0.0.1:5000/register,返回register字符串。
为了项目开发调试方便，前期不需要使用命令行启动flask项目，在manage.py文件中还是使用app.run(),右键运行flask项目。
在操作数据库的语法上和之前使用sqlalchemy还是一样，只不过统一使用db对象去操作数据库。使用起来更方便，更具有了统一性。