python网站开发框架：
django：大而全
flask：小而精
flask的web服务器：werkzeug
模板语法： jinjia2，兼容dtl
登录案例：
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/',methods=['GET'])
def detail(nid):
user = session.get('user_info')
if not user:
return redirect('/login')
info = USERS.get(nid)
return render_template('detail.html',info=info)
@app.route('/index',methods=['GET'])
def index():
user = session.get('user_info')
if not user:
# return redirect('/login')
url = url_for('l1')
return redirect(url)
return render_template('index.html',user_dict=USERS)
@app.route('/login',methods=['GET','POST'],endpoint='l1')
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':
session['user_info'] = user
return redirect('http://www.baidu.com')
return render_template('login.html',error='用户名或密码错误')
if __name__ == '__main__':
app.run()
1 flask 配置文件
django ---》settings.py
flask --->支持很多方式
flask默认主要的一些配置(了解：是否是调试模式，秘钥，cookie的可以值，过期时间),自己的配置(mysql,redis。。)
{
'DEBUG': get_debug_flag(default=False), 是否开启Debug模式
'TESTING': False, 是否开启测试模式
'PROPAGATE_EXCEPTIONS': None,
'PRESERVE_CONTEXT_ON_EXCEPTION': None,
'SECRET_KEY': None,
'PERMANENT_SESSION_LIFETIME': timedelta(days=31),
'USE_X_SENDFILE': False,
'LOGGER_NAME': None,
'LOGGER_HANDLER_POLICY': 'always',
'SERVER_NAME': None,
'APPLICATION_ROOT': None,
'SESSION_COOKIE_NAME': 'session',
'SESSION_COOKIE_DOMAIN': None,
'SESSION_COOKIE_PATH': None,
'SESSION_COOKIE_HTTPONLY': True,
'SESSION_COOKIE_SECURE': False,
'SESSION_REFRESH_EACH_REQUEST': True,
'MAX_CONTENT_LENGTH': None,
'SEND_FILE_MAX_AGE_DEFAULT': timedelta(hours=12),
'TRAP_BAD_REQUEST_ERRORS': False,
'TRAP_HTTP_EXCEPTIONS': False,
'EXPLAIN_TEMPLATE_LOADING': False,
'PREFERRED_URL_SCHEME': 'http',
'JSON_AS_ASCII': True,
'JSON_SORT_KEYS': True,
'JSONIFY_PRETTYPRINT_REGULAR': True,
'JSONIFY_MIMETYPE': 'application/json',
'TEMPLATES_AUTO_RELOAD': None,
}
使用方式--》加载和使用配置
from flask import Flask,jsonify
app=Flask(__name__)
方式一：直接配置--->能够配的参数少
app.secret_key='asdfasdf'
app.debug=True # 修改了代码，只需要保存，自动热加载
方式二：通过app.config字典，配置，这个字典中放了所有的配置
print('---',app.config)
app.debug=True
print(app.config)
app.config['DEBUG']=False # 都要大写
app.config['MYSQL_HOST']='127.0.0.1'
方式三：通过settings.py 配置文件--->用得少，django的方式
app.config.from_pyfile("settings.py")
print(app.config)
方式四：多套配置文件：开发环境，测试环境，线上环境 ，配置有差别
app.config.from_object("settings.DevelopmentConfig")
app.config.from_object("settings.ProductionConfig")
print(app.config)
方式五：服务(项目)多了，配置文件多了---》配置中心 nacos 阿波罗
m={}
m=request.get('ssss')
app.config.from_mapping(m)
@app.route('/')
def index():
# 面试题：你知道的http的请求头和响应头？
'''
referer,user-agent,content-type,cookie-->Connection: keep-alive,X-Forwarded-For
http协议版本号：0.9版本，1.1 版本，2.0版本早就出了
http基于socket---》应用层协议---》osi7层， 5层，tcp/ip 4层--》socket 抽象层--》网络和传输层
三次握手四次挥手，每次发送http，都会三次握手四次挥手-->性能不高
多次http请求共用一个socket连接
2.0 多路复用 socket流式协议
content-type,cookie,
'''
# 方式一:响应头中，响应编码方式为 application/json
return jsonify({'name':"lqz",'age':19})
# 方式二
# res = {'name': "lqz", 'age': 18}
# import json
# res=json.dumps(res)
# return res
if __name__ == '__main__':
app.run()
2 flask 路由系统
你在什么地方用了装饰器，怎么用的---》登陆认证装饰器---》日志装饰器，只要执行这个函数就记录日志
django的路由 urls.py中---》flask中路由基于装饰器
app.router()--->本质是self.add_url_rule，self就是flask对象app
注册路由两种方式
-装饰器
-app.add_url_rule('/',view_func=index)
cbv
如果继承的是View,需要重写dispatch
如果继承的是MethodView，只需要写get，post。。方法即可
class HomeView(MethodView):
def get(self):
print(request.path)
return 'cbv的homeview'
添加路由
name 是路由别名，跟endpoint一个作用，但是cbv必须传name
app.add_url_rule('/home',view_func=HomeView.as_view(name='home'))
app.add_url_rule的参数
'''
1 rule, URL规则, 可以使用转换器
2 endpoint, 当前路由的别名，如果不传， 默认已函数名作为endpoint，如果函数名重名，就会有两个重名的地址，报错,主要用来反向解析
endpoint = None, 名称，用于反向生成URL，即： url_for('名称')
多个视图函数，如果加同一个装饰器，如果不写endpoint，就会报错
3 view_func, 视图函数名称 如果是cbv 视图类.as_view(name='xx')
4 defaults = None, 默认值, 当URL中无参数，函数需要参数时，使用defaults = {'k': 'v'}
为函数提供参数,就是django中的kwargs
5 methods = None, 允许的请求方式，如：["GET", "POST"]
6 strict_slashes = None
对URL最后的 / 符号是否严格要求
7 redirect_to = None, redirect_to='/home'
#重定向到指定地址
8 子域名访问
subdomain = None,
'''
3 flask模板语法
原来dtl中学的，拿过来，无缝衔接---》flask的模板语法支持 （）,[] 等
渲染变量 ---》比dtl多支持允许 () []
for循环一样
if判断，一样
Markup等价django的mark_safe ,
extends,include一模一样
from flask import Flask,render_template,Markup
from flask.views import View,MethodView
app=Flask(__name__)
app.secret_key='asdfasdf'
app.debug=True
def test(a,b):
return a+b
def func1(arg): # Jinja处理了xss攻击，让字符串显示成标签的样子。Markup
return Markup("" %(arg,))
@app.route('/')
def index():
return render_template('index.html',name='lqz',test=test,safe=func1)
if __name__ == '__main__':
app.run()
4 请求与相应
# http请求中有的东西，都能取出来
print(request.method)
# request.method 提交的方法
# request.args get请求提及的数据
print(request.args)
# request.form post请求提交的数据
print(request.form)
# request.values post和get提交的数据总和
print(request.values)
# request.cookies 客户端所带的cookie
print(request.cookies)
# request.headers 请求头
print(request.headers)
# request.path 不带域名，请求路径
print(request.path)
# request.full_path 不带域名，带参数的请求路径
print(request.full_path)
# request.url 带域名带参数的请求路径
print(request.url)
# request.base_url 带域名请求路径
# request.url_root 域名
# request.host_url 域名
# request.host 127.0.0.1:500
print(request.host)
# request.files
# obj = request.files['the_file_name']
# obj.save('/var/www/uploads/' + secure_filename(f.filename))
#1 4件套
# return "字符串"
# return render_template('html模板路径',**{})
# return redirect('/index.html')
# return jsonify({'k1':'v1'})
#1 响应头中加东西,四件套都可以使用make_response包裹成响应对象，等同于django中的HTTPResponse
# res=make_response('字符串')
# res.headers['name']='lqz'
# return res
#2 设置cookie
response = make_response('字符串')
response.set_cookie('key', 'value')
return response
5 session
cookie session token
session是存在于服务端的键值对---》django中的session默认存在数据库的django_session表
request.SESSION['name']='lqz'
#在django中发什么三件事--->响应走，在中间件的process_response中写的
1，生成一个随机的字符串 sdfads
2 往数据库存表
id key content expire
1 sdfads 数据内容(加密) 过期时间
3 写入cookie返回浏览器
response.set_cookie('sessionid',sdfads)
# 请求来了---》process_request中执行了
1 根据sessionid取出随机字符串
2 根据随机字符串去数据库查出content数据，解密
3 赋值给请求对象request.SESSION
4 你在视图函数中才能正常使用request.SESSION['name']取值，赋值，删除值
任何web框讲的session原理都是这个
flask中使用session
设置值：session['login']=True
取值： session['login']
flask中session原理
#flask流程，写入session流程
1 把sesion对象，当字典 转成字符串，使用秘钥加密
val = self.get_signing_serializer(app).dumps(dict(session))
2 写入cookie返回浏览器 session=加密的字符串
response.set_cookie(
app.session_cookie_name,
val, # 加密字符串
)
# 请求来了流程
1 根据sessionid取出加密字符串
val = request.cookies.get(app.session_cookie_name)
2 通过秘钥解密，组装成 session
data = s.loads(val, max_age=max_age)
self.session_class(data)
4 你在视图函数中才能正常使用session['name']取值，赋值，删除值
源码分析
-处理session，有个一个类SecureCookieSessionInterface()，有俩重要方法
-open_session:请求来了执行
1 根据sessionid取出加密字符串
val = request.cookies.get(app.session_cookie_name)
2 通过秘钥解密，组装成 session
data = s.loads(val, max_age=max_age)
self.session_class(data)
4 你在视图函数中才能正常使用session['name']取值，赋值，删除值
-save_session：请求走了执行
1 把sesion对象，当字典 转成字符串，使用秘钥加密
val = self.get_signing_serializer(app).dumps(dict(session))
2 写入cookie返回浏览器 session=加密的字符串
response.set_cookie(
app.session_cookie_name,
val, # 加密字符串
)
# 自定义一个类SecureCookieSessionInterface，重写open_session和save_session，把session存到数据库，redis里----》flask-session 模块做的：数据库，redis，文件中
6 闪现(flash翻译过来的)
在多次请求中，第一次请求放一个值，第二次请求取出这个值，取完就没有了
django中有吗？ 有 没有学---》消息框架，message ，django默认自带的app，message的app
实际用途
-a页面出了错，重定向到b页面，b页面要把错误信息显示
本质：
如果在同一次请求中，放到request对象中即可
如果在不同请求中，放到session中，所以使用闪现一定配置秘钥
使用
设置：flash('诱惑美女')
获取：res=get_flashed_messages()
高级使用 按分类设置和获取
设置：
flash('诱惑美女',category='man')
flash('诱惑帅哥',category='wonmen')
获取：
res=get_flashed_messages(with_categories=True，category_filter=["man"])
1 请求扩展
在请求来了，请求走了，可以做一些校验和拦截，通过装饰器来实现 7 个
1 before_request
类比django中间件中的process_request，在请求收到之前绑定一个函数做一些事情
2 after_request
3 before_first_request
4 teardown_request
5 errorhandler
6 template_global
7 template_filter
from flask import Flask,request,render_template
app=Flask(__name__)
app.debug=True
请求扩展之before_request,在进入视图函数之前执行，如果有多个，是从上往下执行
@app.before_request
def before():
# 判断是否登录，如果登录了继续往下走，如果没登陆
# request.xxx='lqz' 可以往请求中放值，后续可以取出来，只针对于当次请求有效
print("我执行了1111")
@app.before_request
def before1():
# 判断是否登录，如果登录了继续往下走，如果没登陆
print("我执行了2222")
请求走了，会执行它，注册多个，按照从下往上的顺序执行
@app.after_request
def after(response):
# 写入cookie，写入响应头。。。处理跨域
print(response)
print("请求走了1111")
return response
@app.after_request
def after2(response):
print(response)
print("请求走了2222")
return response
before_first_request 只会执行一次，以后就不执行了，跟用户无关，做一些初始化的工作
@app.before_first_request
def first():
print("我的第一次")
#teardown_request 每一个请求之后绑定一个函数，即使遇到了异常，debug为false的情况下
@app.teardown_request
def teardown(e):
print(e) # 记录日志，错误日志request.method
print("我是teardown")
errorhandler 绑定响应状态码，当状态码一致，就会执行它
@app.errorhandler(404)
def err_404(arg):
print(arg)
return render_template('404.html')
@app.errorhandler(500) # 500错误的返回
def err_500(arg):
print(arg)
return render_template('500.html')
标签template_global
@app.template_global()
def add(a1, a2):
return a1 + a2
过滤器template_filter
@app.template_filter()
def add2(a1, a2, a3):
return a1 + a2 + a3
@app.route('/')
def index():
# raise Exception("错误")
l=[1,2,4]
print(l[9])
print("我是视图函数")
return "hello web"
@app.route("/home")
def home():
return render_template('home.html')
if __name__ == '__main__':
app.run()
2 蓝图
blueprint--->现在项目都是单个py文件--》想划分目录--》自己定制目录结构--》app经常用---》各个文件中导来导去---》出现循环引入问题
一个项目有多个app，每个app有自己一套路由，使用蓝图来管理
第一步：定义蓝图对象
user = Blueprint('user', __name__)
第二步：使用蓝图写路径，写请求扩展(只针对于当前蓝图生效)
@user.route('/index')
第三步：把蓝图注册进app中
app.register_blueprint(user)
小型项目目录划分
pro_flask
-pro_flask
-__init__.py
-statics
-templates
-views
account.py
user.py
blog.py
-run.py
# 大型项目 --》有多个app
pro_flask
-pro_flask # 包
__init__.py # 包的init
admin # 包
__init__.py # 包的init
static # 自己的静态文件
templates# 自己的模板文件
models.py
views.py # 自己的视图函数
web # 包
__init__.py
static
templates
views.py
-run.py # 启动文件
以后如何划分，千奇百怪，但是都使用蓝图来做，蓝图管理自己的路由和请求扩展
3 g对象
global的缩写，为了实现在同一个请求中，传递数据,上下文传递，赋值取值只针对于当次请求生效
类比 django 中的 request对象，内有context 属性
django中间件中，假设想放到request中一个数据
request.name="lqz"
rerquest.context['name']='lqz'
rerquest.context['method']='nb'
到视图函数中，就可以取出来
# g和session有什么区别？
-g只针对于当次请求
-session针对于多次请求
1 请求上下文简单分析
flask 1.0带你看的源码
flask的请求上下文-->整个执行过程中有个context对象---》整个贯串了一次请求
1 当请求来的时候，把当次请求的请求对象request，放到local对象中，每个请求都是一个线程在执行
-假设同时来了3个请求，都在执行index视图函数---》3条线程在执行index
-三条线程分别放到local中，只要在当前线程中，再去local中取，取出来的就是当前线程的request
2 视图函数中
-使用request.method，我们发现，全局的request没有乱，不像djagno中一个请求一个request（多线程时只有一个request会乱）
-视图函数中因为同一次请求，包括请求扩展中使用，在同一个线程(协程)中，所以使用的是当次请求的request
-print(request)--->
-print(session)--->
-它俩是同一个类的对象吗？是LocalProxy本地代理类的对象，但是打印出来不一样
-LocalProxy原理是---》重写了所有魔法方法---》代理类---》23种设计模式的--->代理模式
print(LocalProxy对象)--》重写了__str__
print(LocalProxy对象.method)--->重写了__getattr__--->去当次请求的request中取出method
3 请求走了，它要把request对象从local上移除
4 request，session，g 都是全局
5 本质源码中 【request，session】---》放到了ctx对象中--》RequestContext对象--》ctx放到了local上 【g,current_app】-->app_ctx-->APPRequest对象
6 请求上下文，应用上下文
请求来了--》http请求---》符合wsgi协议的web服务器(负责拆分http请求，保装成python对象--》environ, start_response)---》框架(django,flask)--》以后需来一个请求，wsgi服务器就起一个线程---》运行可调用对象加括号---》可调用对象在flask中是 app对象，在django application对象--》对象()---》触发对象的__all__方法---》
flask--->app对象--》__all__是一个请求来了的入口--->
23种设计模式---》架构设计--》封装框架---》大量采用设计模式
-单例模式
-工厂模式
-代理模式
解释了为什么flask中的request是全局对象，而且在不同视图函数中用，没有乱，用的是自己的
2 flask-session(第三方)
缓存到redis，文件。。。。
安装：pip3 install flask-session
本质就是重写了 open_session和save_session
使用方式一，做如下配置，后面跟之前一样
from flask_session import RedisSessionInterface
from redis import Redis
conn = Redis(host='127.0.0.1', port=6379)
app.session_interface = RedisSessionInterface(conn, "session_")
使用方式二：常用的
方式二：通过某个类(app)形式
from redis import Redis
from flask_session import Session
在配置文件中加入数据---》配置文件中配置
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = Redis(host='127.0.0.1',port='6379')
Session(app) # 本质原理跟上面一样，替换了app的session_interface---》更通用--》以后使用flask会经常看到
某个类(app)
#1 设置session过期时间
# cookie默认超时时间是多少？如何设置超时时间
配置文件中 31天
# session过期时间呢？
配置文件配置，PERMANENT_SESSION_LIFETIME，有默认值
2 设置cookie时，如何设定关闭浏览器则cookie失效
如果exipres=None，表示关闭浏览器就失效
response.set_cookie('k','v',exipres=None)#这样设置即可
#在session中设置---》关闭浏览器---》让cookie失效---》只要失效，下次发请求，就不会带来了--》session就没用了
app.session_interface=RedisSessionInterface(conn,key_prefix='lqz',permanent=False)
3 数据库连接池
flask 项目---》使用pymysql操作数据库
直接使用pymysql会存在的问题
-问题1 连接放在全局----》并发安全
-问题2 连接放在视图函数中---》连接数过多
连接池解决
-pip3 install dbutils
-新建一个py文件 pool.py
from dbutils.pooled_db import PooledDB
import pymysql
MYSQL_POOL = PooledDB(
creator=pymysql, # 使用链接数据库的模块
maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数
mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
maxcached=5, # 链接池中最多闲置的链接，0和None不限制
maxshared=3,
# 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制
setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
ping=0,
# ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
host='127.0.0.1',
port=3306,
user='root',
password='123',
database='lqz',
charset='utf8'
)
-在视图函数中使用（只要涉及到多线程，尽量使用池）
from pool import MYSQL_POOL
conn=MYSQL_POOL.connection()
cursor=conn.cursor()
# uwsgi是 进程(4个)+线程模型(开了很多线程)---》4*线程数---》如果io多，性能没有充分利用
# uwsgi+gevent：进程+线程+协程部署方案
# qps：1s内能处理请求数
5 信号
信号--》signal---》flask，django
Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为
整个flask执行过程中，留了一些钩子，可以给钩子绑定函数，一旦绑定了，执行到这，就会执行这个函数
pip3 install blinker
内置信号(flask有，django也有)
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在其中添加数据时，自动触发
信号的使用
信号能干啥：记录日志，当用户访问home，记录日志---》好处是解耦
记录每一个请求的ip地址和客户端类型
django中，只要在user表插入一条数据，我就干个xx事---》双写一致性更新缓存
### 内置信号的使用步骤
信号使用步骤
第一步：定义信号（内置的已经定义好了）
第二步：给信号绑定函数(咱们做)
def func1(*args,**kwargs):
print(args)
print(kwargs)
print("我是信号绑定的函数，我执行了")
before_render_template.connect(func1)
第三步：触发信号(内置信号的触发，在源码中早就写了)
自定义信号的使用
自定义信号
第一步：定义信号
自定义信号
from flask.signals import _signals
lqz = _signals.signal('lqz')
第二步：给信号绑定函数(咱们做)
def func1(*args,**kwargs):
print(args)
print(kwargs)
print("我是信号绑定的函数，我执行了")
lqz.connect(func1)
第三步：触发信号
lqz.send()
6 flask-script
django 可以通过python manage.py runserver 命令来完成一些操作，启动程序，迁移数据。。。
flask不支持--》借助第三方，让它支持---（flask最终变成了djagno）
pip3 install flask-script
django中自定义命令
from flask import Flask, request, session, make_response,render_template
from flask_session import RedisSessionInterface
import pymysql
app = Flask(__name__)
第一步：导入，把app用一个类包一下
from flask_script import Manager
manager=Manager(app)
app.secret_key = 'sss'
django也可以---》自行百度---》项目测试，数据导入--》命令--》自动导入测试数据
python manage.py inportexcel user 路径/a.xsl
自定制命令 python manage.py db init 初始化数据库
### 方式一
@manager.command # python manage.py custom 3 4
def custom(a,b):
print(a)
print(b)
### 方式二
@manager.option('-n', '--name', dest='name')
@manager.option('-u', '--url', dest='url')
def cmd(name, url):
"""
自定义命令（-n也可以写成--name）
执行： python manage.py cmd -n lqz -u www.liuqingzheng.top
执行： python manage.py cmd --name lqz --url www.liuqingzheng.top
"""
print(name, url)
@app.route('/')
def index():
return render_template('index.html')
if __name__ == '__main__':
# app.run()
# 第二步：
manager.run()
1 sqlalchemy介绍和快速使用
SQLAlchemy是一个基于Python实现的ORM框架
django的orm框架---》只能再django中用，不能单独用
SQLAlchemy单独的，可以集成到任意框架中
peewee：轻量级
python的异步orm框架不多， sanic， fastapi---》一旦用了异步，后续所有都需要用异步---》操作mysql，aiomysql--》操作redis，使用aioredis
公司选择
-第一：peewee-async
-第二：框架是异步---》没有使用异步orm框架---》SQLAlchemy---》生成和迁移表---》查询操作数据用原生操作
写django项目---》库和表已经有了
-正常操作django中建表模型---》迁移---》表
-反向生成models--》表---》models.py----》改表---》再反向生成
python manage.py inspectdb > app/models.py
执行原生sql快速使用
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
第一步：创建engine
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/lqz?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
第二步：使用
def task():
conn=engine.raw_connection() # 从连接池中取一个连接
cursor=conn.cursor()
sql="select * from signer"
cursor.execute(sql)
print(cursor.fetchall())
if __name__ == '__main__':
for i in range(20):
t=threading.Thread(target=task)
t.start()
查询mysql的客户端连接数
2 单表操作增删查改
不能创建数据库（django orm也不能）
只能做表的创建和删除，不能做表更改（django orm能)---》借助于第三方实现
###### # 第一步：生成基类，所有表模型都要继承这个基类
django 的orm继承一个父类，Base就是那个父类
###### 第二步：写表模型，继承父类，写字段 （注意区别于django 的orm）
django的default--》可不可以传个函数内存地址---》插入的时候通过函数运算完得到的值
###### 第三步：迁移，通过表模型，生成表
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
第一步：生成基类，所有表模型都要继承这个基类
django 的orm继承一个父类，Base就是那个父类
Base = declarative_base()
第二步：写表模型，继承父类，写字段 （注意区别于django 的orm）
django的default--》可不可以传个函数内存地址---》插入的时候通过函数运算完得到的值
class Users(Base):
id = Column(Integer, primary_key=True,autoincrement=True) # id 主键
name = Column(String(32), index=True, nullable=False) # name列，索引，不可为空
# email = Column(String(32), unique=True) # 唯一
#datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
# ctime = Column(DateTime, default=datetime.datetime.now) # 默认值
# extra = Column(Text, nullable=True) # 大文本，可以为空
__tablename__ = 'lqz_users' # 数据库表名称,如果不写，就报错
__table_args__ = (
# UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一
# Index('ix_id_name', 'name', 'email'), # 联合索引
)
聚簇索(mysql主键自动建索引，聚簇索引，mysql基于聚簇索引构建的B+树)，一定会有，没有显示建主键，mysql会隐藏一个
辅助索引：手动建的叫辅助索引---》单独减了索引---》如果你的辅助索引过多，非常影响插入效率，适度建索引
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 models import Base
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
def create_table():
# 通过engine这个连接配置，创建出所有使用Base管理的表
Base.metadata.create_all(engine)
def delete_table():
# 通过engine这个连接配置，删除出所有使用Base管理的表
Base.metadata.drop_all(engine)
if __name__ == '__main__':
# create_table()
delete_table()
操作表,增加一条记录，以后都用conn/session操作
第一步：创建engin
第二步：通过session得到连接对象
Session = sessionmaker(bind=engine)
session = Session()
# 第三步：实例化得到模型类的对象，增加到数据库中
usr=Users(name='lqz001')
session.add(usr)
# 第四步：提交事务
session.commit()
以后操作数据，都用session对象---》定义在flask的函数外部还是内部？
放内部没问题，每次都生成一个新的session，耗费资源
如果定义在函数外部，会存在 多线程并发使用同一个变量session，要把session做成并发安全的
Session = sessionmaker(bind=engine)
session = scoped_session(Session) # 也是基于local，给每一个线程自己创造一个session
只需要记住，如果是多线程使用，或者在web框架中，使用scoped_session生成session就可以了
集成到flask中，有flask-sqlalchemy第三方，内部已经处理了scoped_session
全局用这个一个session，不用担心并发不安全
usr=Users(name='lqz002')
session.add(usr) # 线程一用：取local中取线程1的那个session，如果就给，没有就重新创造一个
第四步：提交事务
session.commit()
线程一用：取local中取线程1的那个session，如果就给，没有就重新创造一个
线程二用：取local中取线程2的那个session，如果就给，没有就重新创造一个
测试：开3个线程，如果定义全局的session，在3个线程中用，session对象应该是同一个
Session = sessionmaker(bind=engine)
session = Session()
session = scoped_session(Session)
def task():
# usr=Users(name='lqz003')
# session.add(usr)
# session.commit()
# print(session.registry.registry.value) #
print(session) #
开3个线程，如果定义scoped_session，在3个线程中用，session对象应该是不是同一个，独有的
if __name__ == '__main__':
for i in range(3):
t=Thread(target=task)
t.start()
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 sessionmaker
from models import Users
from sqlalchemy.orm import scoped_session
from threading import Thread
第三步：迁移，通过表模型，生成表
from models import Base
from sqlalchemy.sql import text
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
def create_table():
# 通过engine这个连接配置，创建出所有使用Base管理的表
Base.metadata.create_all(engine)
def delete_table():
# 通过engine这个连接配置，删除出所有使用Base管理的表
Base.metadata.drop_all(engine)
if __name__ == '__main__':
# create_table()
# delete_table()
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
#1 增加操作
# obj1 = Users(name="lqz003")
# session.add(obj1)
# # 增加多个，不同对象
# session.add_all([
# Users(name="lqz009"),
# Users(name="lqz008"),
# ])
# session.commit()
# 2 删除操作---》查出来再删---》
# session.query(Users).filter(Users.id > 2).delete()
# session.commit()
# 3 修改操作--》查出来改
# 传字典
# session.query(Users).filter(Users.id > 0).update({"name": "lqz"})
# 类似于django的F查询
# 字符串加
# session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)
# # 数字加
# # session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
# session.commit()
# 4 查询操作----》
# r1 = session.query(Users).all() # 查询所有
# 只取age列，把name重命名为xx
# select name as xx,age from user;
# r2 = session.query(Users.name.label('xx'), Users.age).all()
# filter传的是表达式，filter_by传的是参数
# r3 = session.query(Users).filter(Users.name == "lqz").all()
# r3 = session.query(Users).filter(Users.id >= 1).all()
# r4 = session.query(Users).filter_by(name='lqz').all()
# r5 = session.query(Users).filter_by(name='lqz').first()
#:value 和:name 相当于占位符，用params传参数
# r6 = session.query(Users).filter(text("id
# 更多查询
# 条件
# select * form user where name =lqz
# ret = session.query(Users).filter_by(name='lqz').all()
# 表达式，and条件连接
# select * from user where id >1 and name = lqz
# ret = session.query(Users).filter(Users.id > 1, Users.name == 'lqz').all()
# select * from user where id between 1,3 and name = lqz
# ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'lqz').all()
# # 注意下划线
# select * from user where id in (1,3,4)
# ret = session.query(Users).filter(Users.id.in_([1, 3, 4])).all()
# # ~非，除。。外
# select * from user where id not in (1,3,4)
# ret = session.query(Users).filter(~Users.id.in_([1, 3, 4])).all()
# # 二次筛选
# ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='lqz'))).all()
from sqlalchemy import and_, or_
#
# # or_包裹的都是or条件，and_包裹的都是and条件
# ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
# ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
# ret = session.query(Users).filter(
# or_(
# Users.id < 2,
# and_(Users.name == 'eric', Users.id > 3),
# Users.extra != ""
# )).all()
#
# # 通配符，以e开头，不以e开头
# ret = session.query(Users).filter(Users.name.like('e%')).all()
# ret = session.query(Users).filter(~Users.name.like('e%')).all()
#
# # 限制，用于分页，区间
# ret = session.query(Users)[1:2]
#
# # 排序，根据name降序排列（从大到小）
# ret = session.query(Users).order_by(Users.id.desc()).all()
# # 第一个条件重复后，再按第二个条件升序排
# ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
#
# # 分组
from sqlalchemy.sql import func
# select * from user group by user.extra;
# ret = session.query(Users).group_by(Users.extra).all()
# # 分组之后取最大id，id之和，最小id
# select max(id),sum(id),min(id) from user group by name ;
# ret = session.query(
# func.max(Users.id),
# func.sum(Users.id),
# func.min(Users.id)).group_by(Users.name).all()
# haviing筛选
# select max(id),sum(id),min(id) from user group by name having min(id)>2;
# ret = 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()
# select max(id),sum(id),min(id) from user where id >=1 group by name having min(id)>2;
# ret = session.query(
# func.max(Users.id),
# func.sum(Users.id),
# func.min(Users.id)).filter(Users.id>=1).group_by(Users.name).having(func.min(Users.id) > 2).all()
#
# 连表（默认用forinkey关联）
# select * from user,favor where user.id=favor.id
# ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
# join表，默认是inner join
# select * from Person inner join favor on person.favor=favor.id;
# ret = session.query(Person).join(Favor).all()
# isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
# ret = session.query(Person).join(Favor, isouter=True).all()
# ret = session.query(Favor).join(Person, isouter=True).all()
# 打印原生sql
# aa = session.query(Person).join(Favor, isouter=True)
# print(aa)
# 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
# select * from person left join favor on person.id=favor.id;
# ret = session.query(Person).join(Favor, Person.id == Favor.id, isouter=True).all()
# 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
# union和union all的区别？
# q1 = session.query(Users.name).filter(Users.id > 2)
# q2 = session.query(Favor.caption).filter(Favor.nid < 2)
# ret = q1.union(q2).all()
#
# q1 = session.query(Users.name).filter(Users.id > 2)
# q2 = session.query(Favor.caption).filter(Favor.nid < 2)
# ret = q1.union_all(q2).all()
# 执行原生sql
# 查询
cursor = session.execute('select * from users')
result = cursor.fetchall()
# 添加
cursor = session.execute('insert into users(name) values(:value)', params={"value": 'lqz'})
session.commit()
print(cursor.lastrowid)
django 中执行原生sql---》原生sql用的较多
ret = models.Author.objects.raw('select * from book where nid>1')
把查回来的数据，直接映射给author对象--》name
print(ret)
for i in ret:
print(i)
print(ret.query)
3 一对多
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用于反向查询 # 正向查询按字段，反向查询按 pers
hobby = relationship('Hobby', backref='pers')
一对多
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 sessionmaker
from models import Users
from sqlalchemy.orm import scoped_session
from threading import Thread
第三步：迁移，通过表模型，生成表
from models import Base
from sqlalchemy.sql import text
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
def create_table():
# 通过engine这个连接配置，创建出所有使用Base管理的表
Base.metadata.create_all(engine)
def delete_table():
# 通过engine这个连接配置，删除出所有使用Base管理的表
Base.metadata.drop_all(engine)
if __name__ == '__main__':
create_table()
# delete_table()
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
from models import Hobby,Person
#1 增加数据
# 方式一
# session.add_all([
# Hobby(caption='乒乓球'),
# Hobby(caption='羽毛球'),
# Person(name='张三', hobby_id=1),
# Person(name='李四', hobby_id=1),
# ])
# session.commit()
# 方式二
# person = Person(name='张九', hobby=Hobby(caption='姑娘'))
# session.add(person)
# # 添加三
# hb = Hobby(caption='保龄球')
# # 反向字段
# hb.pers = [Person(name='lqz01'), Person(name='lqz02')]
# session.add(hb)
# session.commit()
#2 查询
# 正向查询
# person = session.query(Person).first()
# print(person.name)
# # 基于对象的跨表查询
# print(person.hobby.caption)
# 反向查询
# v = session.query(Hobby).first()
# print(v.caption)
# print(v.pers) # 多条
# 链表查询
# select person.name ,hobby.caption from person left join bobby on person.hobby_id=hobby.id;
# person_list=session.query(Person.name,Hobby.caption).join(Hobby,isouter=True).all()
# # person_list = session.query(Person,Hobby).join(Hobby, isouter=True).all()
# for row in person_list:
# # print(row.name,row.caption)
# print(row[0].name, row[1].caption)
# person_list = session.query(Person).all()
# for row in person_list:
# print(row.name, row.hobby.caption)
#
# obj = session.query(Hobby).filter(Hobby.id == 1).first()
# persons = obj.pers
# print(persons)
# session.close()
4 多对多
boy girl 相亲，一个boy可以约多个女生，一个女生可以相多个男生
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)
name = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
girls = relationship('Girl', secondary='boy2girl', backref='boys')
多对多
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 sessionmaker
from models import Users
from sqlalchemy.orm import scoped_session
from threading import Thread
第三步：迁移，通过表模型，生成表
from models import Base
from sqlalchemy.sql import text
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
def create_table():
# 通过engine这个连接配置，创建出所有使用Base管理的表
Base.metadata.create_all(engine)
def delete_table():
# 通过engine这个连接配置，删除出所有使用Base管理的表
Base.metadata.drop_all(engine)
from models import Boy,Girl,Boy2Girl
if __name__ == '__main__':
create_table()
# delete_table()
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
# 1 增加数据
## 方式一
# session.add_all([
# Boy(name='彭于晏'),
# Boy(name='刘德华'),
# Girl(name='刘亦菲'),
# Girl(name='迪丽热巴'),
# ])
# session.commit()
# s2g = Boy2Girl(boy_id=1, girl_id=1)
# session.add(s2g)
# session.commit()
## 方式二
# boy = Boy(name='lqz')
# boy.girls = [Girl(name='小红'), Girl(name='校花')]
# session.add(boy)
# session.commit()
### 方式三
# girl = Girl(name='小梅')
# girl.boys = [Boy(name='lqz001'), Boy(name='lqz002')]
# session.add(girl)
# session.commit()
# 基于对象的跨表查
# 使用relationship正向查询
# v = session.query(Boy).first()
# print(v.name)
# print(v.girls)
# 使用relationship反向查询
v = session.query(Girl).first()
print(v.name)
print(v.boys)
5 flask集成
Flask_SQLAlchemy 操作数据库
flask_migrate 模拟django的表迁移
pip3 install flask_migrate
flask_migrate使用步骤
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy() # 全局SQLAlchemy
app = Flask(__name__)
app.config.from_object('settings.DevelopmentConfig')
将db注册到app中，加载配置文件，flask-session，用一个类包裹一下app
db.init_app(app)
flask_script创建命令 runserver命令 ，自定义名字
下面三句会创建出两个命令:runserver db 命令(flask_migrate)
manager=Manager(app)
Migrate(app, db)
manager.add_command('db',MigrateCommand ) # 添加一个db命令，原来有了runserver命令了
直接使用命令迁移表即可
1 python3 manage.py db init #初始化，刚开始干，生成一个migrate文件夹
2 创建表，修改表
python3 manage.py db migrate 等同于 makemigartions
python3 manage.py db upgrade 等同于migrate
Flask_SQLAlchemy给你包装了基类，和session，以后拿到db
db = SQLAlchemy() # 全局SQLAlchemy
增删查改数据-->并发安全
db.session.query()
表模型要继承基表
class Users(db.Model):
Original: https://www.cnblogs.com/elijah-li/p/16398831.html
Author: elijah_li
Title: flask的使用
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/611418/
转载文章受原作者版权保护。转载请注明原作者出处！