Flask
首先,学习flask之前,先了解一下Django和Flask中是怎么渲染字符串的。在Django中使用mark_safe();在Flask中使用Markup();还可以在HTML中使用管道符{{ add | safe }}。简单介绍一下Flask,它是轻量型的,第三方组件非常丰富,可扩展性比较强,用法和Django差不多。
一、介绍Flask、Django、Tornado框架
Django:重武器,内部包含了非常多的组件(ORM、Form、ModelForm、缓存、Session、中间件、信号等)
Flask:短小精悍,内部没有太多的组件,第三方组件非常丰富。路由比较特殊,基于装饰器来实现,但是究其本质还是通过add_url_rule来实现。
Tornado:异步非阻塞框架(node.js)
bottle:第三方库比较少
web.py:比较老
二、Flask入门
安装
pip3 install flask
WSGI
1 from werkzeug.wrappers import Request, Response 2 3 @Request.application 4 def hello(request): 5 return Response('Hello World!') 6 7 if __name__ == '__main__': 8 from werkzeug.serving import run_simple 9 run_simple('localhost', 4000, hello) # 看到run_simple要知道socket就来了 10
werkzeug示例(flask)
from wsgiref.simple_server import make_server def runserver(environ, start_response): start_response('200 OK', [('Content-Type', 'text/html')]) return [bytes('<h1>Hello, web!</h1>', encoding='utf-8'), ] if __name__ == '__main__': # obj = WSGIHandler() # httpd = make_server('', 8000, obj) httpd = make_server('', 8000, runserver) print("Serving HTTP on port 8000...") httpd.serve_forever()
wsgiref示例(wsgi)
1 import socket 2 3 def handle_request(client): 4 buf = client.recv(1024) 5 client.send("HTTP/1.1 200 OK\r\n\r\n") 6 client.send("Hello, Seven") 7 8 def main(): 9 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 10 sock.bind(('localhost',8000)) 11 sock.listen(5) 12 13 while True: 14 connection, address = sock.accept() 15 handle_request(connection) 16 connection.close() 17 18 if __name__ == '__main__': 19 main()
本质的本质
flask
1 from flask import Flask 2 3 # 实例化Flask对象 4 app = Flask(__name__) 5 6 @app.route('/') # -->1.v = app.route('/') 2.v(hello_world) 7 def hello_world(): 8 return 'Hello World!' 9 10 if __name__ == '__main__': 11 app.run()
View Code
三、配置文件
1 class Config(object): 2 DEBUG = False 3 TESTING = False 4 DATABASE_URI = 'sqlite://:memory:' 5 6 class ProductionConfig(Config): 7 DATABASE_URI = 'mysql://user@localhost/foo' 8 9 class DevelopmentConfig(Config): 10 DEBUG = True 11 12 class TestingConfig(Config): 13 TESTING = True 14 15 # 在create_app函数中写上下面这句话,就可以使用配置了 16 app.config.from_object("settings.py.DevelopmentConfig")
settings.py
四、路由系统
设置URL路由使用route()装饰器,route()装饰哪个函数,那么route()中的参数就映射到哪个函数
路由路径支持变量
形式:<converter:variable_name>
converter支持的类型:
1 @app.route('/user/<username>') 2 @app.route('/post/<int:post_id>') 3 @app.route('/post/<float:post_id>') 4 @app.route('/post/<path:path>') 5 @app.route('/login', methods=['GET', 'POST'])
示例
@app.route和@app.add_url_rule参数
1 rule URL规则 2 view_func 视图函数名称 3 defaults=None 默认值,当URL中无参数,函数需要参数时,使用defaults={'k':'v'}为函数提供参数 4 endpoint=None, 名称,用于反向生成URL,即: url_for('名称') 5 methods=None, 允许的请求方式,如:["GET","POST"] 6 strict_slashes=False/True 对URL最后的 / 符号是否严格要求 7 注意:如果设置为True,URL后面没有加"/",访问的时候一定不能加"/" 8 redirect_to=None 重定向到指定地址 9 subdomain=None 子域名访问
View Code
五、模板语言
Flask使用的是jinja2模板,所以其语法和Django无差别
Flask中自定义模板方法的方式和Bottle相似,创建一个函数并通过参数的形式传入render_template
六、请求&响应相关
1 request.method 2 request.args 3 request.form 4 request.values 5 request.cookies 6 request.headers 7 request.path 8 request.full_path 9 request.script_root 10 request.url 11 request.base_url 12 request.url_root 13 request.host_url 14 request.host 15 request.files 16 obj = request.files['the_file_name'] 17 obj.save('/var/www/uploads/' + secure_filename(f.filename))
请求相关信息
1 return "字符串" 2 return render_template('html模板路径',**{}) 3 return redirect('/index.html') 4 5 response = make_response(render_template('index.html')) 6 # response是flask.wrappers.Response类型 7 response.delete_cookie('key') 8 response.set_cookie('key', 'value') 9 response.headers['X-Something'] = 'A value' 10 return response
响应相关信息
七、Session&Cookie
1 from flask import Flask, flash, get_flashed_messages 2 3 app = Flask(__name__) 4 app.secret_key = 'asdfasdf' 5 6 @app.route('/get') 7 def get(): 8 # 从某个地方获取设置过得所有值,并清除 9 data = get_flashed_messages() 10 print(data) 11 return 'Hello World!' 12 13 @app.route('/set') 14 def set(): 15 # 向某个地方设置一个值 16 flash('阿斯蒂芬') 17 return 'Hello World!' 18 if __name__ == '__main__': 19 app.run()
代码示例
九、蓝图(blueprint)
1 @app.before_request 2 def process_request(*args, **kwargs): 3 if request.path == '/login': 4 return None 5 user = session.get('user_info') 6 if user: 7 return None 8 return redirect('/login')
代码示例
执行顺序
1 from flask import Flask, render_template, request, redirect, session, url_for 2 app = Flask(__name__) 3 app.secret_key = 'asdfasdf' 4 5 @app.before_request 6 def process_request1(*args, **kwargs): 7 print('process_request1进来了') 8 9 @app.before_request 10 def process_request2(*args, **kwargs): 11 print('process_request2进来了') 12 13 @app.after_request 14 def process_response1(response): 15 print('process_response1走了') 16 return response 17 18 @app.after_request 19 def process_response2(response): 20 print('process_response2走了') 21 return response 22 23 @app.route('/index', methods=['GET']) 24 def index(): 25 print('index函数') 26 return 'Index' 27 28 if __name__ == '__main__': 29 app.run() 30 31 运行结果: 32 process_request1进来了 33 process_request2进来了 34 index函数 35 process_response2走了 36 process_response1走了
代码示例
请求拦截后,response所有都执行,函数不再执行
1 from flask import Flask, render_template, request, redirect, session, url_for 2 3 app = Flask(__name__) 4 app.secret_key = 'asdfasdf' 5 6 @app.before_request 7 def process_request1(*args, **kwargs): 8 print('process_request1进来了') 9 return '拦截' 10 11 @app.before_request 12 def process_request2(*args, **kwargs): 13 print('process_request2进来了') 14 15 @app.after_request 16 def process_response1(response): 17 print('process_response1走了') 18 return response 19 20 @app.after_request 21 def process_response2(response): 22 print('process_response2走了') 23 return response 24 25 @app.route('/index', methods=['GET']) 26 def index(): 27 print('index函数') 28 return 'Index' 29 30 if __name__ == '__main__': 31 app.run()
代码示例
定制错误信息
1 @app.errorhandler(404) 2 def error_404(arg): 3 return '404错误了'
代码示例
模板中定义方法
1 @app.template_global() 2 def sb(a1, a2): 3 return a1 + a2 4 5 # 在HTNL中使用:{{sb(1,2)}} 6 7 @app.template_filter() 8 def db(a1, a2, a3): 9 return a1 + a2 + a3 10 11 # 在HTNL中使用:{{ 1|db(2,3)}}
代码示例
第一次进来要执行的操作
1 @app.before_first_request 2 def first(*args, **kwargs): 3 pass
代码示例
十一、中间件
1 from flask import Flask, request 2 3 app = Flask(__name__) 4 5 @app.route('/') 6 def index(): 7 return 'Hello World!' 8 9 10 class Md(object): 11 def __init__(self, old_wsgi_app): 12 self.old_wsgi_app = old_wsgi_app 13 def __call__(self, environ, start_response): 14 print('开始之前') 15 ret = self.old_wsgi_app(environ, start_response) 16 print('结束之后') 17 return ret 18 19 if __name__ == '__main__': 20 app.wsgi_app = Md(app.wsgi_app) 21 app.run()
代码示例
十二、上下文管理
1 import threading 2 import time 3 4 local_values = threading.local() 5 6 class Foo(object): 7 def __init__(self): 8 self.name = 0 9 10 local_values = Foo() 11 12 def func(num): 13 local_values.name = num 14 time.sleep(1) 15 print(local_values.name, threading.current_thread().name) 16 17 for i in range(20): 18 th = threading.Thread(target=func, args=(i,), name='线程%s' % i) 19 th.start()
代码示例
源码(request)
1 import threading 2 3 try: 4 from greenlet import getcurrent as get_ident # 协程 5 except ImportError: 6 try: 7 from thread import get_ident 8 except ImportError: 9 from _thread import get_ident # 线程 10 11 12 class local(object): 13 def __init__(self): 14 self.storage = {} 15 self.get_ident = get_ident 16 17 def set(self, k, v): 18 ident = self.get_ident() 19 origin = self.storage.get(ident) 20 if not origin: 21 origin = {k:v} 22 else: 23 origin[k] = v 24 self.storage[ident] = origin 25 26 def get(self, k): 27 ident = self.get_ident() 28 origin = self.storage.get(ident) 29 if not origin: 30 return None 31 return origin.get(k, None) 32 33 local_values = local() 34 35 def task(num): 36 local_values.set('name', num) 37 import time 38 time.sleep(1) 39 print(local_values.get('name'),threading.current_thread().name) 40 41 for i in range(20): 42 th = threading.Thread(target=task, args=(i,), name='线程%s' % i) 43 th.start()
代码示例
自定义类似threading.local对象
1 # ctx = RequestContext对象 2 # 将请求相关的数据environ封装到了RequestContext对象中 3 # 再将对象封装到Local中(每个线程/协程独立空间存储) 4 # ctx.app(当前app的名称) 5 # ctx.request(封装请求相关的东西) 6 # ctx.session 空 7 _request_ctx_stack.local = { 8 唯一标识:{ 9 'stack':[ctx,] 10 }, 11 唯一标识:{ 12 'stack':[ctx,] 13 }, 14 } 15 # app_ctx = AppContext对象 16 # app_ctx.app 17 # app_ctx.g 18 _app_ctx_stack.local = { 19 唯一标识:{ 20 'stack':[app_ctx,] 21 }, 22 唯一标识:{ 23 'stack':[app_ctx,] 24 }, 25 }
View Code
2.使用
1 from flask import request, session, g, current_app 2 print(request, session, g, current_app) 3 都会执行相应LocalProxy对象的__str__ 4 request = LocalProxy(partial(_lookup_req_object, 'request')) 5 session = LocalProxy(partial(_lookup_req_object, 'session')) 6 current_app = LocalProxy(_find_app) 7 g = LocalProxy(partial(_lookup_app_object, 'g'))
View Code
3.终止,ctx、app_ctx全部pop
问题1:多线程是如何实现?
不管几个线程进来都是两个local对象,只不过是每个线程的唯一标识不同,而所有线程的唯一标识都放在对应的Local对象中,使用时取自己对应的不会出错
问题2:flask的local中保存数据时,使用列表创建出来的是栈
如果写web程序,web运行环境:栈中永远保存1条数据(可以不用栈)
写脚本获取app信息时,可能存在app上下文嵌套关系(要使用栈)
1 from flask import Flask, current_app, globals, _app_ctx_stack 2 app1 = Flask('app01') 3 app1.debug = True 4 app2 = Flask('app02') 5 app2.debug = False 6 with app1.app_context(): 7 print(current_app.name) 8 print(_app_ctx_stack._local.__storage__) 9 with app2.app_context(): 10 print(current_app.name) 11 print(_app_ctx_stack._local.__storage__) 12 print(current_app.name)
代码示例
多app应用
1 from werkzeug.wsgi import DispatcherMiddleware 2 from werkzeug.serving import run_simple 3 from flask import Flask, current_app 4 app1 = Flask('app01') 5 app2 = Flask('app02') 6 @app1.route('/index') 7 def index(): 8 print(current_app) 9 return "app01" 10 @app2.route('/index2') 11 def index2(): 12 print(current_app) 13 return "app02" 14 15 # http://www.oldboyedu.com/index 16 # http://www.oldboyedu.com/sec/index2 17 # 通过URL分发 18 dm = DispatcherMiddleware(app1, { 19 '/sec': app2, 20 }) 21 22 if __name__ == '__main__': 23 # app2.__call__ 24 run_simple('localhost', 5000, dm)
代码示例
问题3:web访问多app应用时,上下文管理是如何实现?
每个app都会调用自己的__call__方法,而且都有自己的唯一标识,并且添加到相应的local对象中,只是对应的app是不一样的,执行过程和多线程实现过程类似
补充:当用脚本写flask时,有可能会出现堆栈
1 from flask import Flask, current_app, globals, _app_ctx_stack 2 app1 = Flask('app01') 3 app1.debug = True # 用户名/密码/邮箱 4 app2 = Flask('app02') 5 app2.debug = False 6 # with AppContext(self): # 执行__enter__方法 7 # app_ctx = AppContext(self) 8 # app_ctx.app 9 # app_ctx.g 10 with app1.app_context():# 执行__enter__方法 -> push -> app_ctx添加到_app_ctx_stack.local 11 # {<greenlet.greenlet object at 0x00000184FEEBCCC0>: {'stack': [<flask.ctx.AppContext object at 0x00000184FEFC5748>]}} 12 print(current_app.name) # app01 13 # print(current_app.config['DEBUG']) 14 print(_app_ctx_stack._local.__storage__) 15 with app2.app_context(): 16 # {<greenlet.greenlet object at 0x00000184FEEBCCC0>: {'stack': [<flask.ctx.AppContext object at 0x00000184FEFC5748>, <flask.ctx.AppContext object at 0x00000184FEFC5860>]}} 17 print(current_app.name) # app02 18 # print(current_app.config['DEBUG']) 19 print(_app_ctx_stack._local.__storage__) 20 print(current_app.name) # app01
View Code
1 import functools 2 3 def func(a1, a2): 4 print(a1, a2) 5 6 new_func = functools.partial(func, 666) 7 new_func(999)
代码示例
十四、面向对象
1 class Foo(object): 2 def __init__(self, num): 3 self.num = num 4 5 def __add__(self, other): 6 data = self.num + other.num 7 return Foo(data) 8 9 obj1 = Foo(1) 10 obj2 = Foo(2) 11 v = obj1 + obj2
代码示例
面向对象私有
1 class Foo(object): 2 def __init__(self): 3 self.name = 'alex' 4 self.__age = 18 5 6 def get_age(self): 7 return self.__age 8 9 obj = Foo() 10 print(obj.name) 11 # 强制获取私有字段 12 print(obj._Foo__age) 13 print(obj.get_age())
代码示例
十五、拼接列表中的值
1 from itertools import chain 2 3 v1 = [11, 22, 33] 4 v2 = [44, 55, 66] 5 new = chain(v1, v2) 6 for item in new: 7 print(item)
实例一
1 from itertools import chain 2 3 def f1(x): 4 return x + 1 5 6 func1_list = [f1, lambda x:x-1] 7 8 def f2(x): 9 return x + 10 10 11 new_fun_list = chain([f2], func1_list) 12 13 for func in new_fun_list: 14 print(func)
实例二
十六、数据库连接池(基于threading.local实现)
友情链接:https://www.cnblogs.com/wupeiqi/articles/8184686.html
Django和Flask使用数据库分析
1 import pymysql 2 CONN = pymysql.connect(host='127.0.0.1', 3 port=3306, 4 user='root', 5 password='123', 6 database='pooldb', 7 charset='utf8') 8 cursor = CONN.cursor() 9 cursor.execute('select * from tb1') 10 result = cursor.fetchall() 11 cursor.close() 12 print(result)
代码示例
问题
1.来一个用户连接一次数据库(把连接数据库的操作放到全局变量中)
2.并发运行时,拿到的数据有可能是错的
3.加锁可以解决信息错误的问题,但是没有并发运行
解决思路
不能为每个用户创建一个连接
创建一定数量的连接池,如果有人来使用时有空的就拿去用,用完再还回来,没有时就等待
使用DBUtils模块
1 import time 2 import pymysql 3 from DBUtils.PooledDB import PooledDB 4 5 POOL = PooledDB( 6 creator=pymysql, # 使用链接数据库的模块 7 maxconnections=6, # 连接池允许的最大连接数,0和None表示不限制连接数 8 mincached=2, # 初始化时,链接池中至少创建的空闲的链接,0表示不创建 9 maxcached=5, # 链接池中最多闲置的链接,0和None不限制 10 maxshared=3, # 链接池中最多共享的链接数量,0和None表示全部共享。PS: 无用,因为pymysql和MySQLdb等模块的 threadsafety都为1,所有值无论设置为多少,_maxcached永远为0,所以永远是所有链接都共享。 11 blocking=True, # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错 12 maxusage=None, # 一个链接最多被重复使用的次数,None表示无限制 13 setsession=[], # 开始会话前执行的命令列表。如:["set datestyle to ...", "set time zone ..."] 14 ping=0, 15 # 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 16 host='127.0.0.1', 17 port=3306, 18 user='root', 19 password='123', 20 database='pooldb', 21 charset='utf8' 22 ) 23 24 class SQLHelper(object): 25 @staticmethod 26 def fetch_one(sql, args): 27 conn = POOL.connection() 28 cursor = conn.cursor() 29 cursor.execute(sql, args) 30 result = cursor.fetchone() 31 conn.close() 32 return result 33 34 @staticmethod 35 def fetch_all(sql, args): 36 conn = POOL.connection() 37 cursor = conn.cursor() 38 cursor.execute(sql, args) 39 result = cursor.fetchall() 40 conn.close() 41 return result 42 43 result = SQLHelper.fetch_one('select * from tb1 where id > %s', [1,]) 44 print(result)
代码示例
1 from flask import Flask, session 2 from flask_session import RedisSessionInterface 3 app = Flask(__name__) 4 app.secret_key = 'asdfasdf' 5 6 # 方式一 7 from redis import Redis 8 conn = Redis() 9 app.session_interface = RedisSessionInterface(conn, key_prefix='__', use_signer=False) 10 11 # 方式二 12 from redis import Redis 13 from flask_session import Session 14 app.config['SESSION_TYPE'] = 'redis' 15 app.config['SESSION_REDIS'] = Redis(host='localhost',port='6379') 16 Session(app) 17 18 @app.route('/') 19 def index(): 20 session['xxx'] = 123 21 return "index" 22 23 if __name__ == '__main__': 24 app.run()
代码示例
1 class Foo: 2 def __init__(self): 3 self.age = 23 4 self.name = 'xiaoming' 5 6 class Bar: 7 def __init__(self): 8 self.xx = 123 9 10 # 把类再封装到一个对象里面 11 class Base: 12 def __init__(self): 13 self.f = Foo() 14 self.x = Bar()
View Code
某个值 + 括号
函数/方法:看谁调用,判断函数或方法
类和对象
特殊的双下划线方法,flask中的LocalProxy里面都使用过
__new__ __init__ __call__ __str__ __setattr__ __getattr__ __setitem__ __enter__ __exit__ __add__
View Code
强制调用私有字段,只能在自己这个类中调用
子类和派生类中都不能调用基类私有字段