Flask

一：web框架Django和Flask本質

socket服務端

wsgi： Web服務網關接口
	- wsgiref			# Django內部內置模塊
	- werkzeug			# Flask安裝完成後，內部默認已經安裝好werkzeug

from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple

@Request.application
def hello(request):
    return Response('Hello World!')

if __name__ == '__main__':
    
    run_simple('localhost', 4000, hello)        # hello是回調方法

werkzeug在Flask中完成web框架本質原理

from wsgiref.simple_server import make_server

def run_server(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/html')])
    return [bytes('<h1>Hello, web!</h1>', encoding='utf-8'), ]
 
 
if __name__ == '__main__':
    httpd = make_server('', 8000, run_server)
    httpd.serve_forever()

wsgiref在Django中完成web框架本質原理

 

二：簡單的Flask

建立Flask s1，生成最簡單的代碼。運行s1.py文件，flask運行成功。

from flask import Flask

app = Flask(__name__)


# 路由映射關係
@app.route('/')
def hello_world():
    return 'Hello World!'


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

s1.py

 

三：配置

 如下配置參數爲app = Flask(__name__)的參數，查看源碼類Flask __init__中可傳的參數

import_name,                         # 就是Flask(__name__)中的__name__，通常寫__name__
static_path=None,                    # 靜態文件路徑，這個即將被廢棄了
static_url_path=None,                  # 靜態前綴：static_url_path = '/sssss'。建立flask時目錄被默認建立爲/static,未配置該參數時，訪問127.0.0.1:5000/static/1.jpg就可訪問/static目錄下的圖片
                                        可是修改配置後直接訪問127.0.0.1:5000/sssss/1.jpg就可訪問/static目錄下的圖片
static_folder='static',                # 靜態文件目錄，建立Flask時目錄/static被默認建立
template_folder='templates',           # 模板路徑，建立Flask時目錄/templates被默認建立。from flask import Flask,render_template    return render_template('hello.html')
instance_path=None,                    # C:\Users\Administrator\PycharmProjects\s133\instance，用的少，默認是路徑，當前目錄 + \instance
instance_relative_config=False,        # 當爲True，會默認去C:\Users\Administrator\PycharmProjects\s133\instance找配置文件。若是爲Flase時，無論它。
root_path=None                         # C:\Users\Administrator\PycharmProjects\s133，當前目錄。默認在當前目錄找配置文件instance_relative_config=True時，
                                        默認去C:\Users\Administrator\PycharmProjects\s133\instance找配置文件

View Code

 如下配置爲flask.config.Config對象（繼承字典）的默認參數

{
    '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),                # session的超時時間
    '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,
}

View Code

app.config['DEBUG'] = True    # 進入調試模式
app.debug = True            # 進入調試模式
app.session_interface        # session的接口
app.config.updata({})

配置方式一：(s1.py中經過操做字典的方式)

第一種：
    去一個.py文件中導入配置,例如flask目錄下建立一個settings.py，與staic目錄同一級別
    s133.py:
        app.config.from_pyfile("settings.py")
    settings.py:
        DEBUG = True


第二種：環境變量中取
    app.config.from_envvar("環境變量名稱"),內部調用from_pyfile方法
    使用：
        test.py:
            import os
            os.environ['xxxxx'] = "settings"    # 或者os.environ['xxxxx'] = "settings.py"，settings加入環境變量
        s133.py:
            app.config.from_envvar("xxxxx")        # 找到settings對象，而後執行第一種app.config.from_pyfile("settings.py")


第三種：
    同第一種方式，建立json.py文件,s133.py中調用from_json方法
·    app.config.from_json("json文件名稱")
    JSON文件名稱，必須是json格式，由於內部會執行json.loads


第四種：字典的格式
    app.config.from_mapping({'DEBUG':True})
    
    
第五種：比較推薦使用的，注意要寫大寫，小寫是導入不成功的。
    app.config.from_object("settings.TestingConfig")
    
    settings.py：
        class Config(object):
            DEBUG = False
            TESTING = False
            DATABASE_URI = 'sqlite://:memory:'

        class ProductionConfig(Config):
            DATABASE_URI = 'mysql://user@localhost/foo'

        class DevelopmentConfig(Config):
            DEBUG = True

        class TestingConfig(Config):
            TESTING = True

配置方式二

 

四：路由

路由使用：

@app.route('/')
def hello_world():
    return 'Hello World!'

方法一：裝飾器方式

def hello_world():
    # 反向生成url
    from flask import url_for
    url = url_for('xxx')        # url此時爲 /             
    return 'Hello World!'
app.add_url_rule('/',view_func=hello_world,endpoint='xxx',methods=["GET","POST"])    # view_func視圖函數；endpoint和django中的name同樣，反向生成url，不加endpoint，endpoint默認值爲視圖函數名

方式二：

url正則匹配：

@app.route('/edit/<int:nid>')
def hello_world(nid):
    return 'Hello World!'

示例

@app.route('/user/<username>')
@app.route('/post/<int:post_id>')
@app.route('/post/<float:post_id>')
@app.route('/post/<path:path>')
@app.route('/login', methods=['GET', 'POST'])

經常使用的路由系統,django支持本身寫正則表達式,flask不支持

DEFAULT_CONVERTERS = {
    'default':          UnicodeConverter,
    'string':           UnicodeConverter,
    'any':              AnyConverter,
    'path':             PathConverter,
    'int':              IntegerConverter,
    'float':            FloatConverter,
    'uuid':             UUIDConverter,
}

全部的路由系統都是基於對應關係來處理

from flask import Flask, views, url_for
from werkzeug.routing import BaseConverter

app = Flask(import_name=__name__)


class RegexConverter(BaseConverter):
    """
    自定義URL匹配正則表達式
    """
    def __init__(self, map, regex):
        super(RegexConverter, self).__init__(map)
        self.regex = regex

    def to_python(self, value):
        """
        路由匹配時，匹配成功後傳遞給視圖函數中參數的值
        :param value: 
        :return: 
        """
        return int(value)

    def to_url(self, value):
        """
        使用url_for反向生成URL時，傳遞的參數通過該方法處理，返回的值用於生成URL中的參數
        :param value: 
        :return: 
        """
        val = super(RegexConverter, self).to_url(value)
        return val+'666'

# 添加到flask中
app.url_map.converters['regex'] = RegexConverter

# 自定義的url正則的使用
@app.route('/index/<regex("\d+"):nid>')
def index(nid):
    print(url_for('index', nid='888'))        # 反向生成url /index/888666/ ,反向生成url以前會先執行to_url方法
    return 'Index'


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

自定製url正則匹配

方法一：
    def auth(func):
        def inner(*args, **kwargs):
            print('before')
            result = func(*args, **kwargs)
            print('after')
            return result
    return inner

    @app.route('/index.html',methods=['GET','POST'],endpoint='index')
    @auth
    def index():
        return 'Index'

方法二：
    def auth(func):
        def inner(*args, **kwargs):
            print('before')
            result = func(*args, **kwargs)
            print('after')
            return result
    return inner        
    
    
    class IndexView(views.MethodView):
        methods = ['GET']
        decorators = [auth, ]            # 執行的裝飾器

        def get(self):
            return 'Index.GET'

        def post(self):
            return 'Index.POST'


    app.add_url_rule('/index', view_func=IndexView.as_view(name='index'))  # name=endpoint

flask中裝飾器的使用

rule,                       URL規則
view_func,                  視圖函數名稱
defaults=None,              默認值,當URL中無參數，函數須要參數時，使用defaults={'nid':9}爲函數提供參數
endpoint=None,              名稱，用於反向生成URL，即： url_for('名稱')
methods=None,               容許的請求方式，如：["GET","POST"]


strict_slashes=None,       
    對URL最後的 / 符號是否嚴格要求，
    如：
        @app.route('/index',strict_slashes=False)，
            訪問 http://www.xx.com/index/ 或 http://www.xx.com/index都可
        @app.route('/index',strict_slashes=True)
            僅訪問 http://www.xx.com/index 
            
redirect_to=None,           
    重定向到指定地址
    如：
        @app.route('/index/<int:nid>', redirect_to='/home/<nid>')    # 請求到來不執行/index/<int:nid>代碼，直接重定向到/home/<nid>
        或
        def func(adapter, nid):
            return "/home/888"
        @app.route('/index/<int:nid>', redirect_to=func)
                                
                                
subdomain=None,             
    子域名訪問
    如：
        from flask import Flask, views, url_for

        app = Flask(import_name=__name__)
        app.config['SERVER_NAME'] = 'xuyaping.com:5000'        # 必須寫，才能支持子域名


        @app.route("/index/", subdomain="admin")                    # 訪問http://admin/xuyaping.com:5000/index/        
        def static_index():
            """Flask supports static subdomains
            This is available at static.your-domain.tld"""
            return "static.your-domain.tld"

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

@app.route和app.add_url_rule參數

 

五：模板

模板的使用
　　Flask使用的是Jinja2模板，因此其語法和Django無差異　

　　不過在django模板中執行函數或方法時，不用加括號就會本身執行，而Flask必須本身加括號纔會執行。

　　flask中的Markup等價django的mark_safe

自定義模板方法
　　建立一個函數並經過參數的形式傳入render_template，如：

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <title></title>
</head>
<body>
    <h1>自定義函數</h1>
    {{xyp()|safe}}

</body>
</html>

html

from flask import Flask,render_template
app = Flask(__name__)
 
 
def index():
    return '<h1>index</h1>'
 
@app.route('/login', methods=['GET', 'POST'])
def login():
    return render_template('login.html', ss=index)
 
app.run()

run.py:

 

六：請求和響應

from flask import Flask
from flask import request
from flask import render_template
from flask import redirect
from flask import make_response

app = Flask(__name__)


@app.route('/login.html', methods=['GET', "POST"])
def login():

    # 請求相關信息
    # request.method
    # request.args                # GET傳的參數
    # request.form                # 表單，POST傳的參數
    # request.values
    # request.cookies
    # request.headers
    # request.path
    # request.full_path
    # request.script_root
    # request.url
    # request.base_url
    # request.url_root
    # request.host_url
    # request.host
    # request.files                # 文件
    # obj = request.files['the_file_name']
    # obj.save('/var/www/uploads/')                # save直接把文件存儲到/var/www/uploads/目錄中了

    # 響應相關信息
    # return "字符串"                                # 至關於django中的Httpresponse
    # return render_template('html模板路徑',**{})    # 至關於django中的render
    # return redirect('/index.html')                # 至關於django中的redirect

    # response = make_response(render_template('index.html'))        # make_response把返回的數據封裝起來，而後就有了delete_cookie、set_cookie、headers方法了
    # response是flask.wrappers.Response類型
    # response.delete_cookie('key')
    # response.set_cookie('key', 'value')
    # response.headers['X-Something'] = 'A value'
    # return response


    return "內容"

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

View Code

 

七：session

flask內置session默認放在加密Cookie中，依賴於session.secret_key	
設置：session['username'] ＝ 'xxx'
刪除：session.pop('username', None)

自定義session及使用

import uuid
import json
from flask.sessions import SessionInterface
from flask.sessions import SessionMixin
from itsdangerous import Signer, BadSignature, want_bytes


class MySession(dict, SessionMixin):
    def __init__(self, initial=None, sid=None):
        self.sid = sid
        self.initial = initial
        super(MySession, self).__init__(initial or ())


    def __setitem__(self, key, value):
        super(MySession, self).__setitem__(key, value)

    def __getitem__(self, item):
        return super(MySession, self).__getitem__(item)

    def __delitem__(self, key):
        super(MySession, self).__delitem__(key)



class MySessionInterface(SessionInterface):
    session_class = MySession
    container = {}

    def __init__(self):
        import redis
        self.redis = redis.Redis()

    def _generate_sid(self):
        return str(uuid.uuid4())

    def _get_signer(self, app):
        if not app.secret_key:
            return None
        return Signer(app.secret_key, salt='flask-session',
                      key_derivation='hmac')

    def open_session(self, app, request):
        """
        程序剛啓動時執行，須要返回一個session對象
        """
        sid = request.cookies.get(app.session_cookie_name)
        if not sid:
            sid = self._generate_sid()
            return self.session_class(sid=sid)

        signer = self._get_signer(app)
        try:
            sid_as_bytes = signer.unsign(sid)
            sid = sid_as_bytes.decode()
        except BadSignature:
            sid = self._generate_sid()
            return self.session_class(sid=sid)

        # session保存在redis中
        # val = self.redis.get(sid)
        # session保存在內存中
        val = self.container.get(sid)

        if val is not None:
            try:
                data = json.loads(val)
                return self.session_class(data, sid=sid)
            except:
                return self.session_class(sid=sid)
        return self.session_class(sid=sid)

    def save_session(self, app, session, response):
        """
        程序結束前執行，能夠保存session中全部的值
        如：
            保存到resit
            寫入到用戶cookie
        """
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)
        httponly = self.get_cookie_httponly(app)
        secure = self.get_cookie_secure(app)
        expires = self.get_expiration_time(app, session)

        val = json.dumps(dict(session))

        # session保存在redis中
        # self.redis.setex(name=session.sid, value=val, time=app.permanent_session_lifetime)
        # session保存在內存中
        self.container.setdefault(session.sid, val)

        session_id = self._get_signer(app).sign(want_bytes(session.sid))

        response.set_cookie(app.session_cookie_name, session_id,
                            expires=expires, httponly=httponly,
                            domain=domain, path=path, secure=secure)

sessions.py

from sessions import MySessionInterface
app.session_interface = MySessionInterface()

使用

或者使用flask-session模塊，配置文件中設置

from flask import Flask
from flask import session
from pro_flask.utils.session import MySessionInterface
app = Flask(__name__)

app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
app.session_interface = MySessionInterface()

@app.route('/login.html', methods=['GET', "POST"])
def login():
    print(session)
    session['user1'] = 'alex'
    session['user2'] = 'alex'
    del session['user2']

    return "內容"

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

使用flask-session

 

八：message

message是一個基於Session實現的用於保存數據的集合，其特色是：使用一次就刪除。

from flask import Flask, flash, redirect, render_template, request, get_flashed_messages

app = Flask(__name__)
app.secret_key = 'some_secret'


@app.route('/')
def index1():
    messages = get_flashed_messages()        # 從session中取，取到就刪掉
    print(messages)
    return "Index1"


@app.route('/set')
def index2():
    v = request.args.get('p')
    flash(v)            # 存儲在session中
    return 'ok'


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

View Code

 

九：擴展：僞中間件

from flask import Flask, flash, request

app = Flask(__name__)
app.secret_key = 'some_secret'
 
@app.route('/index')
def index():
    return 'index.html'
 
# 中間件
class MiddleWare:
    def __init__(self,wsgi_app):
        self.wsgi_app = wsgi_app
 
    def __call__(self, environ, start_response):        #  environ, start_response是wsgi socket傳的參數
        print('before')
        response = self.wsgi_app(environ, start_response)
        print('after')
        return response
 
if __name__ == "__main__":
    app.wsgi_app = MiddleWare(app.wsgi_app)
    app.run(port=9999)

View Code

 

十：Flask插件

WTForms    　　    form組件，作form表單驗證的組件
SQLAchemy　　    ORM操做
Flask-Session　　 session插件

　　

十一：藍圖

藍圖的功能就是將不一樣功能放在不一樣的py文件中

eg：

order.py

from flask import Blueprint

order = Blueprint('order',__name__)
@order.route('/order')
def order():
    return 'Order'

account.py

from flask import Blueprint,render_template

account = Blueprint('account',__name__)
@account.route('/login')
def login():
    return render_template('login.html')

_init_.py

from flask import Flask

from .views import account
from .views import order
app = Flask(__name__)
app.register_blueprint(account.account)
app.register_blueprint(order.order)

　　

 十二：數據庫鏈接池

 

"""
爲每一個線程建立一個鏈接，thread.local實現。


"""

from DBUtils.PersistentDB import PersistentDB
import pymysql

POOL = PersistentDB(
    creator=pymysql,  # 使用連接數據庫的模塊
    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
    closeable=False,
    # 若是爲False時， conn.close() 實際上被忽略，供下次使用，再線程關閉時，纔會自動關閉連接。若是爲True時， conn.close()則關閉連接，那麼再次調用pool.connection時就會報錯，由於已經真的關閉了鏈接（pool.steady_connection()能夠獲取一個新的連接）
    threadlocal=None,  # 本線程獨享值得對象，用於保存連接對象，若是連接對象被重置
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='pooldb',
    charset='utf8'
)


def func():
    # conn = SteadyDBConnection()
    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    cursor.close()
    conn.close() # 不是真的關閉，而是假的關閉。 conn = pymysql.connect()   conn.close()

    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    cursor.close()
    conn.close()

import threading

for i in range(10):
    t = threading.Thread(target=func)
    t.start()

模式一

import time
import pymysql
import threading
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='123',
    database='pooldb',
    charset='utf8'
)


def func():
    # 檢測當前正在運行鏈接數的是否小於最大連接數，若是不小於則：等待或報raise TooManyConnections異常
    # 不然
    # 則優先去初始化時建立的連接中獲取連接 SteadyDBConnection。
    # 而後將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中並返回。
    # 若是最開始建立的連接沒有連接，則去建立一個SteadyDBConnection對象，再封裝到PooledDedicatedDBConnection中並返回。
    # 一旦關閉連接後，鏈接就返回到鏈接池讓後續線程繼續使用。

    # PooledDedicatedDBConnection
    conn = POOL.connection()

    # print(th, '連接被拿走了', conn1._con)
    # print(th, '池子裏目前有', pool._idle_cache, '\r\n')

    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    conn.close()

    conn = POOL.connection()
    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    conn.close()

func()

模式二