Flask(二)

一.threading.local

多個線程修改同一個數據，複製多份變量給每一個線程用,爲每一個線程開闢一塊空間進行數據存儲

不用threading.local (多線程資源共享,sleep後執行取到的都是最後一次修改的值)

# 不用local
from threading import Thread
import time
cxw = -1
def task(arg):
    global cxw
    cxw = arg
    # time.sleep(2)
    print(cxw)
​
for i in range(10):
    t = Thread(target=task,args=(i,))
    t.start()

###threading.local使用 (給每個線程都指定了一個id存儲各自修改的屬性值,彼此數據隔離)

from threading import Thread
from threading import local
import time
from threading import get_ident
# 特殊的對象
cxw = local()
def task(arg):
    # 對象.val = 1/2/3/4/5
    cxw.value = arg
    time.sleep(2)
    print(cxw.value)
for i in range(10):
    t = Thread(target=task,args=(i,))
    t.start()

經過字典自定義threading.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()

 

面向對象版

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()

 

經過setattr和getattr實現

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()

 

每一個對象有本身的存儲空間（字典）

from threading import get_ident,Thread
import time
class Local(object):
    def __init__(self):
        object.__setattr__(self,'storage',{})
        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()

 

兼容線程和協程（源碼到request中去看，看local的getattr,setattr）

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()   

partial偏函數

#偏函數的第二個部分(可變參數)，按原有函數的參數順序進行補充，參數將做用在原函數上，最後偏函數返回一個新函數
from functools import partial
def test(a,b,c,d):
    return a+b+c+d
​
tes=partial(test,1,2)
print(tes(3,4))

二.請求上下文

'''
全局的變量
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
current_app = LocalProxy(_find_app)
#local就是咱們的partial(_lookup_req_object, "request")
request = LocalProxy(partial(_lookup_req_object, "request"))
session = LocalProxy(partial(_lookup_req_object, "session"))
g = LocalProxy(partial(_lookup_app_object, "g"))

self.wsgi_app(environ, start_response)源碼：
    def wsgi_app(self, environ, start_response):
        #ctx是ResquestContext的對象，裏面request
        ctx = self.request_context(environ)
        error = None
        try:
            try:
                #就是ctx放到了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
            ctx.auto_pop(error)
1 ctx = self.request_context(environ) environ,請求相關的，ctx如今是包含request,session的RequestContext的對象
    源碼：
        1.1RequestContext(self, environ) self ,是app對象 environ,請求相關的
        1.2 RequestContext在實例化的時候的源碼：
        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._implicit_app_ctx_stack = []
            self.preserved = False
            self._after_request_functions = []
        這個RequestContext對象封裝了,request 和seesoin

2  ctx.push()這個ctx是RequestContext，那就執行RequestContext.push方法
    2.1RequestContext.push()的源碼
        def push(self):
            #_request_ctx_stack是localStack的對象
            #self是ctx,把self也就ctx放入到local對象裏面
            _request_ctx_stack.push(self)
            if self.session is None:
                session_interface = self.app.session_interface
                self.session = session_interface.open_session(self.app, self.request)

                if self.session is None:
                    self.session = session_interface.make_null_session(self.app)

            if self.url_adapter is not None:
                self.match_request()
        2.1.1 _request_ctx_stack.push(self)如今的self是ctx
        2.1.2 _request_ctx_stack是LocalStack()的對象
        2.1.3 LocalStack()的push把ctx傳過來
        2.1.4 LocalStack()的push方法
            源碼：
            #obj是ctx
            def push(self, obj):
                #obj是ctx,requestContext的對象
                rv = getattr(self._local, "stack", None)
                if rv is None:
                    # self._local是Local()的對象
                    # storage[「線程id或者協程id」][stack] = [ctx,]
                    self._local.stack = rv = []
                rv.append(obj)
                return rv

    2的最終也就是ctx.push()他的最終目的：把當前的ctx放入到Local()裏面

3  response = self.full_dispatch_request()
    源碼:
        def full_dispatch_request(self):
            #這是服務器第一次請求時候執行的函數
            self.try_trigger_before_first_request_functions()
            try:
                request_started.send(self)
                #執行請求以前全部的函數，而且拿到請求以前的返回值
                rv = self.preprocess_request()
                if rv is None:
                    #這個是真正視圖函數，若是個人請求以前函數沒有返回值纔會執行
                    rv = self.dispatch_request()
            except Exception as e:
                rv = self.handle_user_exception(e)
            return self.finalize_request(rv)
    3.1 return self.finalize_request(rv)的源碼：
        def finalize_request(self, rv, from_error_handler=False):
            response = self.make_response(rv)
            try:
                #請求以後的函數，after_request
                response = self.process_response(response)
                request_finished.send(self, response=response)
            except Exception:
                if not from_error_handler:
                    raise
                self.logger.exception(
                    "Request finalizing failed with an error while handling an error"
                )
            return response
4 咱們的如今已經在2步的時候把咱們request已經方法Locald對象中了，咱們第三步的任意一個地方
都能使用咱們的request,session,那他是怎麼獲取的？
    4.1 咱們在flask導入request,這個request是一個全局的變量，咱們怎麼經過request區分我當前的request對象(environ)
     咱們發現request是LocalProxy的對象
    4.2 當咱們用全局的request.屬性的時候，就會去找LocalProxy的對象，可是咱們發現裏面根本就沒有
        那他必定執行LocalProxy對象的__getattr__方法
    4.3 咱們如今來看LocalProxy對象的__getattr__方法的源碼：
        #name咱們要獲取屬性名
        def __getattr__(self, name):
            if name == "__members__":
                return dir(self._get_current_object())
            #form
            #self._get_current_object()就是ctx裏面的request，
            return getattr(self._get_current_object(), name)
        4.3.1 經過反射self._get_current_object()對象，來找咱們屬性，也就是name
           self._get_current_object()的源碼：
                def _get_current_object(self):
                    if not hasattr(self.__local, "__release_local__"):
                        return self.__local()
                    try:
                        #self.__local就實例化傳過來的偏函數,
                        return getattr(self.__local, self.__name__)
                    except AttributeError:
                        raise RuntimeError("no object bound to %s" % self.__name__)

           4.3.1.1 return getattr(self.__local, self.__name__)那這裏self.__local是誰？
                    def __init__(self, local, name=None):
                          object.__setattr__(self, "_LocalProxy__local", local)
                    self.___local爲local
                    這個local爲實例化的時候傳的
                4.3.1.1.1 這個實例化的時候的操做
                   request = LocalProxy(partial(_lookup_req_object, "request"))
                   4.3.1.1的local就是 partial(_lookup_req_object, "request")的地址
                4.3.1.1.2 _lookup_req_object的源碼：
                    #調用的時候 partial(_lookup_req_object, "request")
                    #如今的name就是"request"
                    def _lookup_req_object(name):
                        # top是當前線程的ctx
                        top = _request_ctx_stack.top
                        if top is None:
                            raise RuntimeError(_request_ctx_err_msg)
                        #找top裏面的request
                        # ctx找request
                        return getattr(top, name)
                4.3.1.1.2 咱們來看這個_request_ctx_stack.top的top方法
                        def top(self):
                            try:
                                return self._local.stack[-1]
                            except (AttributeError, IndexError):
                                return None
                        咱們發現這個self._local是Local()對象,這樣就把ctx拿到了

'''

 

三.藍圖

對程序進行目錄結構劃分

不使用藍圖，本身分文件

目錄結構：

-templates
-views
 -__init__.py
    -user.py
    -order.py
-app.py

app.py

from views import app
if __name__ == '__main__':
    app.run()

init.py

from flask import Flask,request
app = Flask(__name__)
#不導入這個不行
from . import account
from . import order
from . import user

user.py

from . import app
@app.route('/user')
def user():
    return 'user'

order.py

from . import app
@app.route('/order')
def order():
    return 'order'

使用藍圖之中小型系統

詳見代碼：pro_flask_簡單應用程序目錄示例.zip

目錄結構：

-flask_pro
 -flask_test
    -__init__.py
    -static
        -templates
        -views
        -order.py
            -user.py
     -manage.py 
        

_init.py

from flask import  Flask
app=Flask(__name__)
from flask_test.views import user
from flask_test.views import order
app.register_blueprint(user.us)
app.register_blueprint(order.ord)

manage.py

from flask_test import  app
if __name__ == '__main__':
    app.run(port=8008)

user.py

from flask import Blueprint
us=Blueprint('user',__name__)
​
@us.route('/login')
def login():
    return 'login'

order.py

from flask import Blueprint
ord=Blueprint('order',__name__)
​
@ord.route('/test')
def test():
    return 'order test'

使用藍圖之大型系統

詳見代碼：pro_flask_大型應用目錄示例.zip

總結：

1 xxx = Blueprint('account', __name__,url_prefix='/xxx') ：藍圖URL前綴，表示url的前綴，在該藍圖下全部url都加前綴

2 xxx = Blueprint('account', name,url_prefix='/xxx',template_folder='tpls')：給當前藍圖單獨使用templates，向上查找，當前找不到，會找總templates

3 藍圖的befort_request，對當前藍圖有效

4 大型項目，能夠模擬出相似於django中app的概念

四.g對象

from flask import Flask,views,session,request,g
app=Flask(__name__)
''''
session 只要設置，在任意請求中都能拿到，不管你拿多少次
flash   一旦設置，可在任意一次請求中獲取，可是隻能取一次
g     一旦設置，只能在當前請求中獲取，其它的請求都不能獲取


'''

@app.after_request
def test(response):
    print(g.name)
    return response



@app.route("/")
def index():
    g.name = "suv"
    return "ok"

@app.route("/index")
def index1():

    return "ojbk"



if __name__ == '__main__':

    app.run()

五.flask-session

做用：將默認保存的簽名cookie中的值 保存到 redis/memcached/file/Mongodb/SQLAlchemy

安裝：pip3 install flask-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)

#use_signer是否對key簽名
#若是use_siginer爲False,這表示不須要配置app.secret_key
#permanent=False表示關閉瀏覽器cookie失效

app.secret_key="aksdhkajs"

app.session_interface=RedisSessionInterface(conn,key_prefix='lqz',
                                            use_signer=True,permanent=False)

'''
以前的session
seesion名字爲配置文件中的名字
存
seesion  ->加密-->cookie
取
session -->值  ----》解密

redis的seesion:
seesion名字爲配置文件中的名字(默認爲session)

前臺cookie =>  session:session.id

redis => name:value
name=self.key_prefix + session.sid, value=val 是redis的值


'''
@app.route('/')
def hello_world():
    session['name']='lqz'
    return 'Hello World!'

@app.route("/index")
def index():
    print(session['name'])
    return "ok"


if __name__ == '__main__':
    app.run()

第二種簡便方式:

from flask import Flask,session
from redis import Redis
from flask_session import Session
app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = Redis(host='127.0.0.1',port='6379')
Session(app)

@app.route('/')
def hello_world():
    session['name']='lqz'
    return 'Hello World!'

@app.route("/index")
def index():
    print(session['name'])
    return "ok"


if __name__ == '__main__':
    app.run()

 

六.信號

Flask框架中的信號基於blinker，其主要就是讓開發者但是在flask請求過程當中定製一些用戶行爲

安裝：pip3 install blinker

內置信號：

```python
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在其中添加數據時，自動觸發
```

使用信號：

from flask import Flask,signals,render_template

app = Flask(__name__)

# 往信號中註冊函數
def func12123(*args,**kwargs):
    print('觸發型號',args,kwargs)
signals.request_started.connect(func12123)

# 觸發信號： signals.request_started.send()
@app.before_first_request
def before_first1(*args,**kwargs):
    pass
@app.before_first_request
def before_first2(*args,**kwargs):
    pass

@app.before_request
def before_first3(*args,**kwargs):
    pass

@app.route('/',methods=['GET',"POST"])
def index():
    print('視圖')
    return "ok"


if __name__ == '__main__':
    app.wsgi_app
    app.run()

自定義信號:

from flask import Flask, current_app, flash, render_template
from flask.signals import _signals

app = Flask(import_name=__name__)

# 自定義信號
xxxxx = _signals.signal('xxxxx')


# 必須有一個位置參數，去接收他的發送者，
def func(sender,name):
    print(sender)
    print(name)
    print("123")



# 自定義信號中註冊函數
xxxxx.connect(func)


@app.route("/x")
def index():
    # 觸發信號，這裏的第一是發送者，第二個參數可選的話，必須是鍵值對
    xxxxx.send("sb",name="wb")
    return 'Index'


if __name__ == '__main__':
    app.run()