首先,學習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
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()
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()
三、配置文件
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")
四、路由系統
設置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 子域名訪問
五、模板語言
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 }
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'))
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
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()
某個值 + 括號
函數/方法:看誰調用,判斷函數或方法
類和對象
特殊的雙下划線方法,flask中的LocalProxy里面都使用過
__new__ __init__ __call__ __str__ __setattr__ __getattr__ __setitem__ __enter__ __exit__ __add__
強制調用私有字段,只能在自己這個類中調用
子類和派生類中都不能調用基類私有字段
