Flask進階

時間  2019-12-07
標籤 flask 進階 欄目 Python 简体版
原文   原文鏈接

Threading.local

　　做用：爲每一個線程建立一個獨立的空間，使得線程對本身的空間中的數據進行操做（數據隔離）。css

　　應用：html

    1. flask上下文管理中的local中比這更高級，爲協程。
      2. DBUtils線程池的模式一：爲每一個線程建立一個鏈接
    2. SQLAchemy
"""
import threading
from threading import local
import time

obj = local()


def task(i):
    obj.xxxxx = i
    time.sleep(2)
    print(obj.xxxxx,i)

for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()
"""
"""
import threading
from threading import local

def task(i):
    print(threading.get_ident(),i)

for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()
"""
"""
import time
import threading
import greenlet

DIC = {}

def task(i):

    # ident = threading.get_ident()
    ident = greenlet.getcurrent()
    if ident in DIC:
        DIC[ident]['xxxxx'] = i
    else:
        DIC[ident] = {'xxxxx':i }
    time.sleep(2)

    print(DIC[ident]['xxxxx'],i)

for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()

"""
import time
import threading


try:
    import greenlet
    get_ident = greenlet.getcurrent
except Exception:
    get_ident = threading.get_ident


class Local(object):
    DIC = {}

    def __getattr__(self, item):
        ident = get_ident()
        if ident in self.DIC:
            return self.DIC[ident].get(item)
        return None

    def __setattr__(self, key, value):
        ident = get_ident()
        if ident in self.DIC:
            self.DIC[ident][key] = value
        else:
            self.DIC[ident] = {key: value}


obj = Local()


def task(i):
    obj.xxxxx = i
    time.sleep(2)
    print(obj.xxxxx, i)


for i in range(10):
    t = threading.Thread(target=task, args=(i, ))
    t.start()
threading.local.py 
- 如何獲取一個線程的惟一標記？ threading.get_ident()
- 根據字典自定義一個相似於threading.local功能？
import time
import threading

DIC = {}

def task(i):
    ident = threading.get_ident()
    if ident in DIC:
        DIC[ident]['xxxxx'] = i
    else:
        DIC[ident] = {'xxxxx':i }
    time.sleep(2)

    print(DIC[ident]['xxxxx'],i)

for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()

- 根據字典自定義一個爲每一個協程開闢空間進行存取數據。

import time
import threading
import greenlet

DIC = {}

def task(i):
    
    # ident = threading.get_ident()
    ident = greenlet.getcurrent()
    if ident in DIC:
        DIC[ident]['xxxxx'] = i
    else:
        DIC[ident] = {'xxxxx':i }
    time.sleep(2)

    print(DIC[ident]['xxxxx'],i)

for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()

- 經過getattr/setattr 構造出來 threading.local的增強版（協程）
import time
import threading
try:
    import greenlet
    get_ident =  greenlet.getcurrent
except Exception as e:
    get_ident = threading.get_ident

class Local(object):
    DIC = {}

    def __getattr__(self, item):
        ident = get_ident()
        if ident in self.DIC:
            return self.DIC[ident].get(item)
        return None

    def __setattr__(self, key, value):
        ident = get_ident()
        if ident in self.DIC:
            self.DIC[ident][key] = value
        else:
            self.DIC[ident] = {key:value}
        

obj = Local()

def task(i):
    obj.xxxxx = i
    time.sleep(2)
    print(obj.xxxxx,i)

for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()
問題

　　應用：DBUtils中爲每一個線程建立一個數據庫鏈接時使用java

  1、上下文管理

兩類請求：請求上下文，app上下文python

上下文管理的實現（重點）mysql

當請求到來的時候，
    flask會把請求相關和session相關信息封裝到一個ctx = RequestContext對象中，
    會把app和g封裝到app_ctx = APPContext對象中去，
    經過localstack對象將ctx、app_ctx對象封裝到local對象中

    問題：
        local是什麼？做用？
            爲每一個線程建立一個獨立的空間，使得線程對本身的空間中的數據進行操做（數據隔離）
        localstack是什麼？做用
            storage = {
                1234:{stack:[ctx,]}
            }
            localstack將local中的stack中的維護成一個棧
獲取數據(執行視圖函數的時候)
    經過導入一個模塊，用模塊.的方式獲取咱們想要的數據
    實現細節：
        經過localproxy對象+偏函數，調用localstack去local中獲取對應的響應ctx、app_ctx中封裝值
        問題：爲何要把ctx = request/session  app_ctx = app/g
            由於離線腳本須要使用app_ctx
請求結束
    調用localstk的pop方法，將ctx和app_ctx移除

1. 請求上下文管理

　　基礎知識web

- 偏函數：給函數取一個默認的參數,這樣可使函數沒必要每一次都傳參。
        應用：request = LocalProxy(partial(_lookup_req_object, 'request'))
            session = LocalProxy(partial(_lookup_req_object, 'session'))
    import functools

    def index(a1,a2):
        return a1 + a2

    # 原來的調用方式
    # ret = index(1,23)
    # print(ret)

    # 偏函數，幫助開發者自動傳遞參數
    new_func = functools.partial(index,666)
    ret = new_func(1)
    print(ret)
偏函數 
- super和執行類的區別？
"""
class Base(object):

    def func(self):
        print('Base.func')

class Foo(Base):

    def func(self):
        # 方式一：根據mro的順序執行方法
        # super(Foo,self).func()
        # 方式二：主動執行Base類的方法
        # Base.func(self)

        print('Foo.func')


obj = Foo()
obj.func()
"""
#################################### 
class Base(object):

    def func(self):
        super(Base, self).func()
        print('Base.func')

class Bar(object):
    def func(self):
        print('Bar.func')

class Foo(Base,Bar):
    pass

# 示例一
# obj = Foo()
# obj.func()
# print(Foo.__mro__)

# 示例二
# obj = Base()
# obj.func()
super和執行類的區別 
# -*- coding: utf-8 -*-
'''
# @Datetime: 2018/12/23
# @author: Zhang Yafei
'''
'''
class Base(object):
    def func(self):
        print('Base_func')


class Foo(Base):
    def func(self):
        # 方式一：根據mro的順序執行方法
        # super(Foo, self).func()
        # 方式二：主動執行Base類的方法
        # Base.func(self)
        print('Foo_func')


obj = Foo()
obj.func()
'''


############################### super和執行父類的方法的區別  ###################
class Base(object):
    def func(self):
        super(Base, self).func()
        print('Base_func')


class Bar(object):
    def func(self):
        print('Bar_func')


class Foo(Base, Bar):
    pass


obj = Foo()
obj.func()
print(Foo.__mro__)

# obj = Base()
print(Base.__mro__)
# obj.func()
執行父類方法 
# -*- coding: utf-8 -*-
'''
# @Datetime: 2018/12/23
# @author: Zhang Yafei
'''


class Foo(object):
    def __init__(self):
        object.__setattr__(self, 'storage', {})

    def __setattr__(self, key, value):
        print(key, value, self.storage)


obj = Foo()
obj.xx = 123
特殊方法_getattr__和__setattr__ 
class Stack(object):
    """棧"""
    def __init__(self):
        self.__list = []  #私有變量，不容許外部調用者對其進行操做

    def push(self,item):
        """添加一個新的元素item到棧頂"""
        self.__list.append(item)   #順序表尾部插入時間複雜度O(1),頭部插入O(n),故尾部方便
        #self.__list.insert(0,item)   #鏈表表尾部插入時間複雜度O(n),頭部插入O(1),故鏈表用頭插方便

    def pop(self):
        """彈出棧頂元素"""
        return self.__list.pop()

if __name__ == '__main__':
    s = Stack()
    s.push(1)
    s.push(2)
    s.push(3)
    s.push(4)

    print(s.pop())
    print(s.pop())
    print(s.pop())
    print(s.pop())
基於列表實現棧 
'''
# @Datetime: 2018/12/25
# @author: Zhang Yafei
'''


class Obj(object):
    def func(self):
        pass


# 執行方法一
# obj = Obj()
# obj.func()    # 方法

# 執行方法二
# obj = Obj()
# Obj.func(obj)   # 函數

from types import FunctionType, MethodType

obj = Obj()
print(isinstance(obj.func, FunctionType))  # False
print(isinstance(obj.func, MethodType))  # True

print(isinstance(Obj.func, FunctionType))  # True
print(isinstance(Obj.func, MethodType))  # Flase
"""
總結：函數和方法的區別
    1.被類調用的是函數，被對象調用的是方法
    2.所有參數本身傳的是函數，自動傳的是方法(好比self)
"""
方法和函數的區別 
# -*- coding: utf-8 -*-

"""
@Datetime: 2018/12/29
@Author: Zhang Yafei
"""
"""方式一"""
# my_singleton.py
#
# class Singleton(object):
#     pass
#
# singleton = Singleton()
#
# from mysingleton import singleton

"""方式二：使用裝飾器"""
# def Singleton(cls):
#     _instance = {}
#
#     def _singleton(*args, **kargs):
#         if cls not in _instance:
#             _instance[cls] = cls(*args, **kargs)
#         return _instance[cls]
#
#     return _singleton
#
#
# @Singleton
# class A(object):
#     a = 1
#
#     def __init__(self, x=0):
#         self.x = x
#
#
# a1 = A(2)
# a2 = A(3)

"""方式三：使用類"""
# import threading
# import time
#
#
# class Singleton(object):
#
#     def __init__(self):
#         # time.sleep(1)
#         pass
#     @classmethod
#     def instance(cls, *args, **kwargs):
#         if not hasattr(Singleton, "_instance"):
#             Singleton._instance = Singleton(*args, **kwargs)
#         return Singleton._instance
#
#
# def task(arg):
#     obj = Singleton.instance()
#     print(obj)
#
#
# for i in range(10):
#     t = threading.Thread(target=task,args=[i,])
#     t.start()

"""解決方法：加鎖"""
# import time
# import threading
#
#
# class Singleton(object):
#     _instance_lock = threading.Lock()
#
#     def __init__(self):
#         time.sleep(1)
#
#     @classmethod
#     def instance(cls, *args, **kwargs):
#         if not hasattr(Singleton, "_instance"):
#             with Singleton._instance_lock:
#                 if not hasattr(Singleton, "_instance"):
#                     Singleton._instance = Singleton(*args, **kwargs)
#         return Singleton._instance
#
#
#
#
# def task(arg):
#     obj = Singleton.instance()
#     print(obj)
#
# for i in range(10):
#     t = threading.Thread(target=task,args=[i,])
#     t.start()
# time.sleep(20)
# obj = Singleton.instance()
# print(obj)

"""方法四：基於__new__方法"""
#
# import threading
#
# class Singleton(object):
#     _instance_lock = threading.Lock()
#
#     def __init__(self):
#         pass
#
#     def __new__(cls, *args, **kwargs):
#         if not hasattr(Singleton, "_instance"):
#             with Singleton._instance_lock:
#                 if not hasattr(Singleton, "_instance"):
#                     Singleton._instance = object.__new__(cls)
#         return Singleton._instance
#
#
# obj1 = Singleton()
# obj2 = Singleton()
# print(obj1,obj2)
#
#
# def task(arg):
#     obj = Singleton()
#     print(obj)
#
#
# for i in range(10):
#     t = threading.Thread(target=task,args=[i,])
#     t.start()
#
"""方法五：元類實現單例模式"""
import threading


class SingletonType(type):
    _instance_lock = threading.Lock()

    def __call__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            with SingletonType._instance_lock:
                if not hasattr(cls, "_instance"):
                    cls._instance = super(SingletonType,cls).__call__(*args, **kwargs)
        return cls._instance


class Foo(metaclass=SingletonType):
    def __init__(self,name):
        self.name = name


obj1 = Foo('name')
obj2 = Foo('name')
print(obj1,obj2)
單例模式

　　執行流程redis

請求到來時候：wsgi會觸發__call__方法,由__call__方法再次調用wsgi_app方法
            -在wsgi_app方法中：
                首先將 請求相關+空session 封裝到一個RequestContext對象中，即ctx
                將ctx交給LocalStack中，再由localstack將ctx添加到local中，local結構：
                    __storage__ = {
                        1231:{stack:[ctx，]}
                    }
                -根據請求中的cookie中提取名稱爲sessionnid對應的值，對cookie進行加密+反序列化，再次賦值給ctx中的session
                -> 視圖函數
                - 把session中的數據再次寫入到cookie中
                -將ctx刪除
            -結果返回用戶瀏覽器
            -斷開socket鏈接

    # ctx = RequestContext(self, environ) # self是app對象，environ請求相關的原始數據
    # ctx.request = Request(environ)
    # ctx.session = None
    
    # 將包含了request/session的ctx對象放到「空調」
        {
            1232：{ctx:ctx對象}
            1231：{ctx:ctx對象}
            1211：{ctx:ctx對象}
            1111：{ctx:ctx對象}
            1261：{ctx:ctx對象}
        }
        
視圖函數：
    from flask import reuqest,session 
    
    request.method 
    
    
請求結束：
    根據當前線程的惟一標記，將「空調」上的數據移除。
請求上下文執行流程

　　主要內容sql

a. 溫大爺：wsig,  
    run_simple() 自動執行__call__方法,__call__方法調用wsgi_app()方法
b.   趙毅：
            ctx = ReuqestContext(session,request) # 將request和session封裝成一個ctx對象
            ctx.push()  # 經過localstack添加到local(空調)中
c.   劉鬆：LocalStack，把ctx對象添加到local中
        LocalStack的做用：線程標記做爲字典的key,字典裏面包含key爲stack的字典，value爲一個列表,
                        localstack的做用就是將這個列表維護成一個stack.
d.   空調：Local
            __storage__={
                1321：{stack:[ctx,]}
            }
上下文管理request 
a. 溫大爺：wsig
b.   趙毅：
            ctx = ReuqestContext(session=None,request)
            ctx.push()
c.   劉鬆：LocalStack，把ctx對象添加到local中
d.   空調：Local
            __storage__={
                1321：{stack:[ctx,]}
            }
e.   郭浩：經過劉鬆獲取ctx中的session，給session賦值（從cookie中讀取數據） => open_session
上下文管理session 
pip3 install flask-session 

掌握：
    - 使用
        import redis
        from flask import Flask,request,session
        from flask.sessions import SecureCookieSessionInterface
        from flask_session import Session

        app = Flask(__name__)

        # app.session_interface = SecureCookieSessionInterface()
        # app.session_interface = RedisSessionInterface()
        app.config['SESSION_TYPE'] = 'redis'
        app.config['SESSION_REDIS'] = redis.Redis(host='140.143.227.206',port=6379,password='1234')
        Session(app)

        @app.route('/login')
        def login():
            session['user'] = 'alex'
            return 'asdfasfd'

        @app.route('/home')
        def index():
            print(session.get('user'))

            return '...'


        if __name__ == '__main__':
            app.run()
    - 原理：
        - session數據保存到redis
            session:隨機字符串1:q23asifaksdfkajsdfasdf
            session:隨機字符串2:q23asifaksdfkajsdfasdf
            session:隨機字符串3:q23asifaksdfkajsdfasdf
            session:隨機字符串4:q23asifaksdfkajsdfasdf
            session:隨機字符串5:q23asifaksdfkajsdfasdf
        - 隨機字符串返回給用戶,瀏覽器。
            隨機字符串
        
        
        源碼：
            from flask_session import RedisSessionInterface
flask-session

2. 應用上下文管理

  • - 請求上下文（ctx=RequestContext()）：request/session
  • - App上下文（app_ctx=AppContext()）: app/g
- 程序啓動：
    兩個Local：
        local1 = {
        
        }
        
        local2 = {
        
        }

    兩個LocalStack:
        _request_ctx_stack
        _app_ctx_stack
- 請求到來
    對數據進行封裝：
        ctx = RequestContext(request,session)
        app_ctx = AppContext(app,g)
    保存數據：
        將包含了(app,g)數據的app_ctx對象，利用 _app_ctx_stack（貝貝，LocalStack()）將app_ctx添加到Local中
            storage = {
                1231:{stack:[app_ctx(app,g),]}
            }
        將包含了request,session數據的ctx對象，利用_request_ctx_stack（劉淞，LocalStack()），將ctx添加到Local中
            storage = {
                1231:{stack:[ctx(request,session),]}

- 視圖函數處理：
    from flask import Flask,request,session,current_app,g

    app = Flask(__name__)


    @app.route('/index')
    def index():
        # 去請求上下文中獲取值 _request_ctx_stack
        request.method # 找小東北獲取值
        session['xxx'] # 找龍泰獲取值
        
        # 去app上下文中獲取值：_app_ctx_stack 
        print(current_app)
        print(g)
        
        return "Index"


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

- 結束
    _app_ctx_stack.pop()
    _request_ctx_stack.pop()

app上下文管理執行流程
應用上下文管理執行流程

3.視圖函數中獲取上下文信息注意點

　　請求上下文和應用上下文須要先放入Local中，才能獲取到數據庫

# -*- coding: utf-8 -*-

"""
@Datetime: 2019/1/4
@Author: Zhang Yafei
"""
from flask import Flask, current_app, request, session, g

app = Flask(__name__)

# 錯誤：程序解釋的時候尚未執行__call__方法，尚未將上下文信息放入local中，因此取不到會報錯
# print(current_app.config)


@app.route('/index')
def index():
    # 正確
    print(current_app.config)
    return 'index'


if __name__ == '__main__':
    app.run()
視圖函數中使用 
# -*- coding: utf-8 -*-

"""
@Datetime: 2018/12/31
@Author: Zhang Yafei
"""
from web import db, create_app
from flask import current_app

# 錯誤，尚未放入local中
# print(current_app.config)  # RuntimeError: Working outside of application context.

app = create_app()     # app_ctx = app/g
app_ctx = app.app_context()

with app_ctx:  # __enter__,經過LocalStack放入Local中
    # db.create_all()   # 調用LocalStack放入Local中獲取app，再去app中獲取配置
    # 正確
    print(current_app.config)
離線腳本中使用

 4.問題總結

1. Flask中g的生命週期？
    (1) 當一個請求到來時，flask會把app和g封裝成一個AppContext對象，
    (2) 經過_app_ctx_stack = LocalStack()(做用是將local維護成一個棧)，將g存放到local中.
        app_ctx = _app_ctx_stack.top
        app_ctx.push()
        _app_ctx_stack.push(self)
        rv = getattr(self._local, 'stack', None)
        rv.append(obj)
    (3) 執行視圖函數：
        g = LocalProxy(partial(_lookup_app_object, 'g'))
        g.x 經過localProxy對象去local中取g
    (4) _app_ctx_stack.pop()  # 將g清空
    - 一次請求聲明週期結束
    注：兩次請求的g不同
2. g和session同樣嗎？
    g只能在同一次請求起做用，請求結束以後就銷燬了，曇花一現
    但session請求完以後不銷燬，不一樣請求均可以使用
3. g和全局變量同樣嗎？
    g是項目啓動的時候建立的，且一次請求結束以後刪除，能夠經過線程標記實現多線程
    全局變量屢次請求不刪除，能夠共用，但不能多線程
問題

2、進階功能擴展

1. 連接和版本管理器:方便對url和版本進行管理

class UrlManager(object):
    @staticmethod
    def buildUrl(path):
        return path

    @staticmethod
    def buildStaticUrl(path):
        path = path + '?ver=' + '201901292105'
        return UrlManager.buildUrl(path)

url = url_for('index')     # /index
url_1 = UrlManager.buildUrl('/api')   # /api
url_2 = UrlManager.buildStaticUrl('/css/bootstrap.css')    # /css/bootstrap.css?ver=201901292105
msg = 'Hello World,url:{} url_1:{} url_2:{}'.format(url, url_1, url_2)   # .....

　2.日誌系統

A logging.Logger object for this application. The default configuration is to log to stderr if the application is in debug mode. This logger can be used to (surprise) log messages. Here some examples:

app.logger.debug('A value for debugging')
app.logger.warning('A warning occurred (%d apples)', 42)
app.logger.error('An error occurred')

　3.錯誤處理：使用戶有更好的使用體驗

@app.errorhandler(404)
def page_not_found(error):
    app.logger.error(error)
    return 'This page does not exist', 404　

3、第三方組件

1. 數據庫鏈接池：DBUtils

from DBUtils.PooledDB import PooledDB, SharedDBConnection

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='0000',
database='flask_code',
charset='utf8'
)

-注意：使用數據庫鏈接池
from settings import POOL
conn = POOL.connection()
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
View Code

2.wtforms

-使用
    生成HTML標籤
    form表單驗證
-安裝
    pip install wtforms
-使用    
    示例：用戶登陸
            用戶註冊
            實時更新：從數據庫獲取字段，靜態字段只在程序掃描第一遍時執行。
                解決方法：1.重啓  2. 重寫form的構造方法，每一次實例化都動態更新須要更新的字段
問題：
    1. form對象爲何能夠被for循環？
        答：變爲可迭代對象：                
            class Obj(object):
                def __iter__(self):
                    # return iter([1,2,3])
                    yield 1
                    yield 2
                    yield 3


            obj = Obj()

            for item in obj:
                print(item)
    2. new方法的返回值決定對象究竟是什麼？
        class BAR(object):
            def __init__(self, cls):
                self.cls = cls


        class NEW_OBJ(object):

            def __new__(cls, *args, **kwargs):
                # return super(NEW_OBJ, cls).__new__(cls, *args, **kwargs)   # <__main__.NEW_OBJ object at 0x000000D445061CF8>
                # return 123     # 123
                # return BAR          # <class '__main__.BAR'>
                # return BAR()        # <__main__.BAR object at 0x000000AD77141C50>
                return BAR(cls)       # <__main__.BAR object at 0x0000003BFFA31D68>


        obj = NEW_OBJ()
        print(obj)
    
    3. metaclass
        1. 建立類時,先執行metaclass(默認爲type)的__init__方法
        2. 類在實例化時, 執行metaclass(默認爲type)的__call__方法，__call__方法的返回值就是實例化的對象
            __call__內部調用：
                - 類.__new__方法：建立對象
                _ 類.__init__方法：對象的初始化

            class MyType(type):
                def __init__(self, *args, **kwargs):
                    print('MyType的__init__')
                    super(MyType, self).__init__(*args, **kwargs)

                def __call__(cls, *args, **kwargs):
                    print('MyType的__call__')
                    obj3 = cls.__new__(cls)
                    cls.__init__(obj3)
                    return obj


            class Obj3(object, metaclass=MyType):
                x = 123

                def __init__(self):
                    print('Obj3的__init__')

                def __new__(cls, *args, **kwargs):
                    print('Obj3的__new__')
                    return object.__new__(cls)

                def func(self):
                    return 666

        源碼：
            -類的建立
                type.__init__
            -對象的建立：
                type.__call__
                    類.__new__
                    類.__init__

Meta
    - __mro__
    - 應用：wtforms中meta使用（meta做用定製csrf token）
鉤子函數
    class LoginForm(Form):
        name = simple.StringField(
            validators=[
                validators.DataRequired(message='用戶名不能爲空.'),
            ],
            widget=widgets.TextInput(),
            render_kw={'placeholder':'請輸入用戶名'}
        )
        pwd = simple.PasswordField(
            validators=[
                validators.DataRequired(message='密碼不能爲空.'),

            ],
            render_kw={'placeholder':'請輸入密碼'}
        )

        def validate_name(self, field):
            """
            自定義name字段規則
            :param field:
            :return:
            """
            # 最開始初始化時，self.data中已經有全部的值
            print('鉤子函數獲取的值',field.data)
            if not field.data.startswith('old'):
                raise validators.ValidationError("用戶名必須以old開頭") # 繼續後續驗證
                # raise validators.StopValidation("用戶名必須以old開頭")  # 再也不繼續後續驗證

總結：
    記住： https://www.cnblogs.com/wupeiqi/articles/8202357.html
    理解：
            - metaclass 
            - __new__ 
            - __mro__
View Code

3.flask-sqlalchemy　　

pip install flask-sqlalchemy
下載安裝 
導入並實例化SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

注意事項：
    - 必須在導入藍圖以前
    - 必須導入models.py
web.__init__.py 
db.init_app(app)
初始化 
# ##### SQLALchemy配置文件 #####
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:0000@127.0.0.1:3306/flask_web?charset=utf8"
SQLALCHEMY_POOL_SIZE = 10
SQLALCHEMY_MAX_OVERFLOW = 5
配置文件配置 
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy import Integer,String,Text,Date,DateTime
from sqlalchemy import create_engine
from chun import db


class Users(db.Model):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    depart_id = Column(Integer)
models.py 
from web import db,create_app

app = create_app()
app_ctx = app.app_context() # app_ctx = app/g
with app_ctx: # __enter__,經過LocalStack放入Local中
    db.create_all() # 調用LocalStack放入Local中獲取app，再去app中獲取配置
生成表 
from flask import Blueprint
from web import db
from web import models
us = Blueprint('us',__name__)


@us.route('/index')
def index():
    # 使用SQLAlchemy在數據庫中插入一條數據
    # db.session.add(models.Users(name='高件套',depart_id=1))
    # db.session.commit()
    # db.session.remove()
    result = db.session.query(models.Users).all()
    print(result)
    db.session.remove()

    return 'Index'
基於orm對數據庫進行操做

4.flask_script

#  python manage.py runserver -h 127.0.0.1 -p 8001

from web import create_app
from flask_script import Manager


app = create_app()
manager = Manager(app)

if __name__ == '__main__':
    # app.run()
    manager.run()
增長runserver 
from web import create_app
from flask_script import Manager


app = create_app()
manager = Manager(app)

@manager.command
def custom(arg):
    """
    自定義命令
    python manage.py custom 123
    :param arg:
    :return:
    """
    print(arg)


if __name__ == '__main__':
    # app.run()
    manager.run()
位置傳參 
from web import create_app
from flask_script import Manager


app = create_app()
manager = Manager(app)

@manager.option('-n', '--name', dest='name')
@manager.option('-u', '--url', dest='url')
def cmd(name, url):
    """
    自定義命令
    執行： python manage.py  cmd -n wupeiqi -u http://www.oldboyedu.com
    :param name:
    :param url:
    :return:
    """
    print(name, url)


if __name__ == '__main__':
    # app.run()
    manager.run()
關鍵字傳參

5. flask-sqlacodegen

　　使用 flask-sqlacodegen 擴展 方便快速生成 ORM model

　　(1) 安裝編程

pip install flask-sqlacodegen

　　(2)使用

flask-sqlacodegen "mysql://root:0000@127.0.0.1/food_db" --outfile "common/models/model.py"  --flask
flask-sqlacodegen "mysql://root:0000@127.0.0.1/food_db" --tables user --outfile "common/models/user.py"  --flask

6. flask-migrate：利用sqlalchemy對數據庫中的表進行遷移操做

　　依賴：flask-script 

from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from web import create_app
from web import db


app = create_app()
manager = Manager(app)
Migrate(app, db)

"""
# 數據庫遷移命名
    python manage.py db init
    python manage.py db migrate
    python manage.py db upgrade
"""
manager.add_command('db', MigrateCommand)


if __name__ == '__main__':
    manager.run()
    # app.run()
manage.py

7.找到項目使用的全部組件及版本

第一步:pip install pipreqs    
第二步：
    進入項目目錄：pipreqs ./ --encoding=utf-8 

從文件中安裝包：pip install -r requirements.txt

補充：
    虛擬環境導出包：pip freeze>requeriments.txt

8.虛擬環境

方式一：pip3 install virtualenv    
        virtualenv env1  --no-site-packages

方式二：:python -m venv mysite_venv
進入script目錄：
    激活虛擬環境：activate
    退出虛擬環境：deactivate

   4、打造高可用的MVC框架

   5、信號

　　信號即就是框架給咱們預留出一些位置，讓咱們能經過幾行簡單的代碼在生命週期的某個位置執行一些操做

　1.內置信號

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

源碼示例

class Flask(_PackageBoundObject):

    def full_dispatch_request(self):
       
        self.try_trigger_before_first_request_functions()
        try:
            # ############### 觸發request_started 信號 ###############
            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)
        response = self.make_response(rv)
        response = self.process_response(response)

        # ############### request_finished 信號 ###############
        request_finished.send(self, response=response)
        return response

    def wsgi_app(self, environ, start_response):
        
        ctx = self.request_context(environ)
        ctx.push()
        error = None
        try:
            try:
                response = self.full_dispatch_request()
            except Exception as e:
                error = e
                response = self.make_response(self.handle_exception(e))
            return response(environ, start_response)
        finally:
            if self.should_ignore_error(error):
                error = None
            ctx.auto_pop(error)
request_started 
def render_template(template_name_or_list, **context):
    """Renders a template from the template folder with the given
    context.

    :param template_name_or_list: the name of the template to be
                                  rendered, or an iterable with template names
                                  the first one existing will be rendered
    :param context: the variables that should be available in the
                    context of the template.
    """
    ctx = _app_ctx_stack.top
    ctx.app.update_template_context(context)
    return _render(ctx.app.jinja_env.get_or_select_template(template_name_or_list),
                   context, ctx.app)

def _render(template, context, app):
    """Renders the template and fires the signal"""

    # ############### before_render_template 信號 ###############
    before_render_template.send(app, template=template, context=context)
    rv = template.render(context)
    
    # ############### template_rendered 信號 ###############
    template_rendered.send(app, template=template, context=context)
    return rv

before_render_template
before_render_template 
class Flask(_PackageBoundObject):

    def handle_exception(self, e):
       
        exc_type, exc_value, tb = sys.exc_info()

        # ############### got_request_exception 信號 ###############
        got_request_exception.send(self, exception=e)
        handler = self._find_error_handler(InternalServerError())

        if self.propagate_exceptions:
            # if we want to repropagate the exception, we can attempt to
            # raise it with the whole traceback in case we can do that
            # (the function was actually called from the except part)
            # otherwise, we just raise the error again
            if exc_value is e:
                reraise(exc_type, exc_value, tb)
            else:
                raise e

        self.log_exception((exc_type, exc_value, tb))
        if handler is None:
            return InternalServerError()
        return handler(e)

    def wsgi_app(self, environ, start_response):
        
        ctx = self.request_context(environ)
        ctx.push()
        error = None
        try:
            try:
                response = self.full_dispatch_request()
            except Exception as e:
                error = e
                # 這裏這裏這裏這裏這裏這裏這裏這裏這裏這裏這裏這裏 #
                response = self.make_response(self.handle_exception(e))
            return response(environ, start_response)
        finally:
            if self.should_ignore_error(error):
                error = None
            ctx.auto_pop(error)

got_request_exception
got_request_exception 
class AppContext(object):
    def push(self):
        """Binds the app context to the current context."""
        self._refcnt += 1
        if hasattr(sys, 'exc_clear'):
            sys.exc_clear()
        _app_ctx_stack.push(self)
        # ############## 觸發 appcontext_pushed 信號 ##############
        appcontext_pushed.send(self.app)

    def pop(self, exc=_sentinel):
        """Pops the app context."""
        try:
            self._refcnt -= 1
            if self._refcnt <= 0:
                if exc is _sentinel:
                    exc = sys.exc_info()[1]
                # ############## 觸發 appcontext_tearing_down 信號 ##############
                self.app.do_teardown_appcontext(exc)
        finally:
            rv = _app_ctx_stack.pop()
        assert rv is self, 'Popped wrong app context.  (%r instead of %r)' \
            % (rv, self)

        # ############## 觸發 appcontext_popped 信號 ##############
        appcontext_popped.send(self.app)

class RequestContext(object):
    def push(self):
        top = _request_ctx_stack.top
        if top is not None and top.preserved:
            top.pop(top._preserved_exc)

        app_ctx = _app_ctx_stack.top
        if app_ctx is None or app_ctx.app != self.app:
            
            # ####################################################
            app_ctx = self.app.app_context()
            app_ctx.push()
            self._implicit_app_ctx_stack.append(app_ctx)
        else:
            self._implicit_app_ctx_stack.append(None)

        if hasattr(sys, 'exc_clear'):
            sys.exc_clear()

        _request_ctx_stack.push(self)

        # Open the session at the moment that the request context is
        # available. This allows a custom open_session method to use the
        # request context (e.g. code that access database information
        # stored on `g` instead of the appcontext).
        self.session = self.app.open_session(self.request)
        if self.session is None:
            self.session = self.app.make_null_session()

class Flask(_PackageBoundObject):


    def wsgi_app(self, environ, start_response):
        
        ctx = self.request_context(environ)
        ctx.push()
        error = None
        try:
            try:
                response = self.full_dispatch_request()
            except Exception as e:
                error = e
                response = self.make_response(self.handle_exception(e))
            return response(environ, start_response)
        finally:
            if self.should_ignore_error(error):
                error = None
            ctx.auto_pop(error)


    def pop(self, exc=_sentinel):
        app_ctx = self._implicit_app_ctx_stack.pop()

        try:
            clear_request = False
            if not self._implicit_app_ctx_stack:
                self.preserved = False
                self._preserved_exc = None
                if exc is _sentinel:
                    exc = sys.exc_info()[1]

                # ################## 觸發 request_tearing_down 信號 ##################
                self.app.do_teardown_request(exc)

                # If this interpreter supports clearing the exception information
                # we do that now.  This will only go into effect on Python 2.x,
                # on 3.x it disappears automatically at the end of the exception
                # stack.
                if hasattr(sys, 'exc_clear'):
                    sys.exc_clear()

                request_close = getattr(self.request, 'close', None)
                if request_close is not None:
                    request_close()
                clear_request = True
        finally:
            rv = _request_ctx_stack.pop()

            # get rid of circular dependencies at the end of the request
            # so that we don't require the GC to be active.
            if clear_request:
                rv.request.environ['werkzeug.request'] = None

            # Get rid of the app as well if necessary.
            if app_ctx is not None:
                # ####################################################
                app_ctx.pop(exc)

            assert rv is self, 'Popped wrong request context.  ' \
                '(%r instead of %r)' % (rv, self)

    def auto_pop(self, exc):
        if self.request.environ.get('flask._preserve_context') or \
           (exc is not None and self.app.preserve_context_on_exception):
            self.preserved = True
            self._preserved_exc = exc
        else:
            self.pop(exc)
request_tearing_down 
def flash(message, category='message'):
    """Flashes a message to the next request.  In order to remove the
    flashed message from the session and to display it to the user,
    the template has to call :func:`get_flashed_messages`.

    .. versionchanged:: 0.3
       `category` parameter added.

    :param message: the message to be flashed.
    :param category: the category for the message.  The following values
                     are recommended: ``'message'`` for any kind of message,
                     ``'error'`` for errors, ``'info'`` for information
                     messages and ``'warning'`` for warnings.  However any
                     kind of string can be used as category.
    """
    # Original implementation:
    #
    #     session.setdefault('_flashes', []).append((category, message))
    #
    # This assumed that changes made to mutable structures in the session are
    # are always in sync with the session object, which is not true for session
    # implementations that use external storage for keeping their keys/values.
    flashes = session.get('_flashes', [])
    flashes.append((category, message))
    session['_flashes'] = flashes

    # ############### 觸發 message_flashed 信號 ###############
    message_flashed.send(current_app._get_current_object(),
                         message=message, category=category)
message_flashed

自定義信號

# -*- coding: utf-8 -*-

"""
@Datetime: 2019/1/7
@Author: Zhang Yafei
"""
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, *args, **kwargs):
    print(sender)


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


@app.route("/x")
def index():
    # 觸發信號
    xxxxx.send('123123', k1='v1')
    return 'Index'


if __name__ == '__main__':
    app.run()
自定義信號

 

重點：

1. 上下文管理
    當請求到來的時候，
        flask會把請求相關和session相關信息封裝到一個ctx = RequestContext對象中，
        會把app和g封裝到app_ctx = APPContext對象中去，
        經過localstack對象將ctx、app_ctx對象封裝到local對象中

        問題：
            local是什麼？做用？
                爲每一個線程建立一個獨立的空間，使得線程對本身的空間中的數據進行操做（數據隔離）
            localstack是什麼？做用
                storage = {
                    1234:{stack:[ctx,]}
                }
                localstack將local中的stack中的維護成一個棧
    獲取數據
        經過localproxy對象+偏函數，調用localstack去local中獲取響應ctx、app_ctx中封裝值
        問題：爲何要把ctx = request/session  app_ctx = app/g
            由於離線腳本須要使用app_ctx
    請求結束
        調用localstk的pop方法，將ctx和app_ctx移除    
2.程序中有多少個local和localstack?
    請求localstack,幫助開發在對stack對應列表操做
        請求local = {
            1212:{
                stack:[ctx,]
            }
        }
    應用localstack,幫助開發在對stack對應列表操做
        應用local = {
            1212：{
                stack:[app_ctx,]
            }
        }
    
3.爲何要建立兩個local和兩個localstack?
    - 編寫離線腳本時，須要配置文件，而配置文件存放在app中。
    - 編寫離線腳本時，不須要請求相關的數據
    
    因此，將app和請求相關的數據分開：
        應用上下文(app,g)
        請求上下文(request,session)

4. Local中做用？
    爲每一個協程建立一個獨立的空間，使得協程對本身的空間中的數據進行操做（數據隔離）
相似於threading.local的做用，可是是他升級版(greentlet.get_currt())
__storage__ =  {
    1231:{},
    1232:{},
}
5.LocalStack做用？
    將local中存儲數據的列表stack維護成一個棧。協程標記做爲字典的key,字典裏面包含key爲stack的字典，
    value爲一個列表,localstack的做用就是將這個列表維護成一個stack.
    __storage__ =  {
                1231:{stack:[],},
                1232:{stack:[],},
    }
6.爲何要維護成一個棧？
    web運行時（web runtime）:
        請求Local = {
            1111:{
                stack:[ctx1,]
            },
            1112:{
                stack:[ctx2,]
            },
            1113:{
                stack:[ctx3,]
            }
        }
        
        應用Local = {
            1111:{
                stack:[app_ctx1,]
            }
            1112:{
                stack:[app_ctx2,]
            },
            1113:{
                stack:[app_ctx3,]
            }
        }
    多app離線腳本：
        from flask import current_app
        app1 = create_app()
        app_ctx1 = app1.app_context() # app_ctx = app/g

        app2 = create_app()
        app_ctx2 = app2.app_context() # app_ctx = app/g

        with app_ctx1: # __enter__,經過LocalStack放入Local中
            print(current_app) # app1
            with app_ctx2:
                print(current_app) # app2
                
            print(current_app) # app1

        """
        請求Local = {
        }

        應用Local = {
            1111:{
                stack:[app_ctx1,app_ctx2]
            }
        }
        """
7. threading.local做用？
    爲每一個線程建立一個獨立的空間，使得線程對本身的空間中的數據進行操做（數據隔離）
8. LocalProxy做用？
    執行視圖函數的時候，LocalProxy經過Localstack從local中取數據
9. g的做用？
    相似每次請求中的全局變量，但要注意的是g只存在於一次請求中，請求結束以後就銷燬了。
10.爲何要導入request,就可使用？
    每次執行request.xx方法時，會觸發LocalProxy對象的__getattr__等方法，由方法每次動態的使用LocalStack去Local中獲取數據。
10. before_request、after_request 實現原理？
    將視圖函數添加到一個列表中，而後循環執行其中的函數，after_request將列表reverse一下。
11. 對面向對象的理解？
    python支持函數式編程和麪向對象編程。java和c#只支持面向對象編程。
    基礎：談面向對象就要從他的三大特性開始提及，如：封裝、繼承、多態。
        封裝：
            - 方法封裝到來類中:某一類功能類似的方法
                class File:
                    def file_add():pass


                    def file_update():pass


                    def file_del():pass


                    def file_fetch():pass
    
            - 數據封裝到對象中
                class File:
                    def __init__(self,name,age,email):
                        self.name = name 
                        self.age = age 
                        self.email = email 
                    def file_add():pass


                    def file_update():pass


                    def file_del():pass


                    def file_fetch():pass

                obj1 = File('oldboy',19,"asdf@live.com")
                obj2 = File('oldboy1',119,"asdf12@live.com")
            
            應用：
                - session/request封裝到了RequestContext對象中
                - app/g封裝到了AppContext中
                
        繼承：若是多個類中有相同的方法，爲了不重複編寫，能夠將其放在父類（基類）中。
            python支持多繼承，繼承順序按__mro__的順序執行，新式類按廣度優先，舊式類類廣度優先，Python3都是新式類。先執行左邊，在執行右邊
            
            class Base(object):
                def xxxx():pass

            class File(Base):
                def __init__(self,name,age,email):
                    self.name = name 
                    self.age = age 
                    self.email = email 
                def file_add():pass


                def file_update():pass


                def file_del():pass


                def file_fetch():pass

            class DB(Base):
                def db_add():pass


                def db_update():pass


                def db_del():pass

                def xxxx():pass
                
                def db_fetch():pass

            應用：    
                rest framework中的視圖類的繼承
                

        多態（鴨子模型）：天生支持多態，對於參數來講能夠傳入任何類型的對象，只要保證有想要的send方法便可。
            
            class Msg(object):
                def send():
                    pass 
                    
            class WX(object):
                def send():
                    pass 
            
        
            def func(arg):
                arg.send()
                
    進階：
        __init__，初始化
        __new__，建立對象
        __call__，對象()
        __getattr__，對象.xx 且xx不存在
        __getattribute__, 對象.xx  xx爲任何屬性
        __setattr__. 對象.xx = yy
        __delattr__, del 對象.xx
        __setitem__，對象['xx'] = yy
        __getitem__, 對象['xx']
        __delitem__, del 對象['xx']
        
        __mro__，查找成員順序
        __str__, 對象返回值
        __repr__，
        __iter__， 實現此方法，且返回一個迭代器，此對象可迭代
        __dict__， 類的成員
        __add__， 類 + xx
        __del__, 對象的生命週期結束以後

    高級：metaclass
        1. 類建立 
            class Foo(object):pass 
            
            Foo = type('Foo',(object,),{})
        2. 如何指定類由自定義type建立？
            class MyType(type):
                pass 
            
            class Foo(object,metaclass=MyType):
                # __metaclass__ = MyType    # py2
                pass 
            
            Foo = MyType('Foo',(object,),{})
        3. 默認執行順序
        
            class Foo(object,metaclass=MyType):
                pass 
                
            obj = Foo()
            
            
            
            class MyType(type):
                def __init__(self,*args,**kwargs):
                    print('111')
                    super(MyType,self).__init__(*args,**kwargs)


            class Base(object, metaclass=MyType):
                pass

            class Foo(Base):
                pass
            
            若是一類本身或基類中指定了metaclass，那麼該類就是由metaclass指定的type或mytype建立。
            
            同：
                class MyType(type):
                    def __init__(self,*args,**kwargs):
                        print('111')
                        super(MyType,self).__init__(*args,**kwargs)


                # class Base(object, metaclass=MyType):
                #     pass

                Base = MyType('Base',(object,),{})

                class Foo(Base):
                    pass
            同：
                class MyType(type):
                    def __init__(self,*args,**kwargs):
                        print('111')
                        super(MyType,self).__init__(*args,**kwargs)


                # class Base(object, metaclass=MyType):
                #     pass
                def with_metaclass(arg):
                    Base = MyType('Base',(arg,),{})
                    return Base

                class Foo(with_metaclass(object)):
                    pass
flask常見問題
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程