flask

Flask是一個基於jinja2模板和Werkzeug Wsgi服務的一個微型框架。
wsgi web服務網關接口，實現該協議的模塊：wsgiref,werkzeug Werkzeug 本質是一個Socket服務端，敢於接受http請求並將請求進行預處理，而後發給Flask框架，
開發人員基於FLask框架提供的功能對請求進行相應的處理，並返回給用戶。
若是要返回給用戶複雜的內容時，須要藉助jinja2模板來實現對模板的處理，：將模板和數據進行渲染，
將渲染後的字符串返回給用戶瀏覽器

Flask小，但擴展性很是強大

from werkzeug.wrappers import Request, Response

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

if __name__ == '__main__':
    from werkzeug.serving import run_simple
    run_simple('localhost', 4000, hello)　　flask在這使用的對象

 與django最大不一樣點  request，session

一、配置文件

flask的配置文件是以個flask.config.Config對象，繼承了字典

import importlib
path = "setting.Foo"
module_name, cls_name = path.rsplit(path, maxsplit=1)
module = importlib.import_module(module_name)
cls = getattr(module, cls_name)
for key in dir(cls):
    if key.isupper():
        print(key, getattr(cls, key))

獲取配置的原理

{
    'DEBUG':                                get_debug_flag(default=False),  是否開啓Debug模式
    'TESTING':                              False,                          是否開啓測試模式
    'PROPAGATE_EXCEPTIONS':                 None,                          
    'PRESERVE_CONTEXT_ON_EXCEPTION':        None,
    'USE_X_SENDFILE':                       False,
    'LOGGER_NAME':                          None,
    'LOGGER_HANDLER_POLICY':               'always',
    'SERVER_NAME':                          None,
    'APPLICATION_ROOT':                     None,
    
    'SECRET_KEY':                           None,
    'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),
    '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,
}

Config默認配置

修改配置信息:

app.config["DEBUG"]=True
由於他是個dict也可使用.update() app.config.from_pyfile("settings.py") # 從配置文件加載配置 　　　　文件格式： 　　　　　　DEBUG = True

app.config.from_envvar("環境變量的名字")    # 從環境變量加載配置
app.config.from_json("json文件的名字")     # 從json文件加載配置
app.config.from_mapping({'DEBUG':True})         #  字典格式
app.config.from_object('pro_flask.settings.TestingConfig')    # 從類獲取

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

一、使用
二、原理

 

應用： importlib、反射、getattr
django中間件
restframwork    全局配置　　APIVIEW

 

二、路由系統

@app.route("/index/<username>",methods=[],endpoint="')

@app.route("/index/<int:post_id>")　　# 類型轉換　　float、path、default、string、any、int、uuid、

# 一旦有<>函數就必須接受這個關鍵字參數，另外還能夠指定請求的參數 methods=["GET"，「POST」]　　

url_for("endpoint",**kwarg)  反向解析，若是路由沒有指定endpoint，默認是裝飾的函數

。。。。。。。。
------->必定要有
import functools def auth(func): @functools.wraps(func)
適用於給較少的函數添加

 

一、先執行 decorator = app.route("/index")
二、@decorator

裝飾器與閉包函數： 裝飾器是閉包函數的一種應用，經過@將函數名傳入閉包函數並返回

@app.route和app.add_url_rule參數：
rule,                       URL規則
view_func,                  視圖函數名稱
defaults=None,              默認值,當URL中無參數，函數須要參數時，使用defaults={'k':'v'}爲函數提供參數
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>')
                                或
                                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'] = 'wupeiqi.com:5000'


                                    @app.route("/", subdomain="admin")
                                    def static_index():
                                        """Flask supports static subdomains
                                        This is available at static.your-domain.tld"""
                                        return "static.your-domain.tld"


                                    @app.route("/dynamic", subdomain="<username>")
                                    def username_index(username):
                                        """Dynamic subdomains are also supported
                                        Try going to user1.your-domain.tld/dynamic"""
                                        return username + ".your-domain.tld"


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

@app.route和app.add_url_rule參數：

 

 

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

app = Flask(__name__)


class RegexConverter(BaseConverter):
    """自定義URL匹配正則表達式"""

    def __init__(self, map, regex):
        super().__init__(map)
        self.regex = regex

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

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


"""
一、用戶發送請求
二、flask內部進行正則匹配
三、調用to_python(正則匹配的結果)方法
四、to_python方法的返回值交給視圖函數的參數
"""
# 添加到flask中
app.url_map.converters['regex'] = RegexConverter


@app.route('/index/<regex("\d+"):nid>')
def index(nid):
    print(url_for('index', nid='888'))
    return 'Index'


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

自定義正則

 

三、視圖

 

註冊視圖
-------------------------------------------------
@app.route("/index/<id>",methods=["get"])
def index(id):
    return render_template('index.html')
--------------------------------------------------
def index():
    return "Index"

app.add_url_rule(rule='/', endpoint="index", view_func=index, methods=["GET", "POST"])
---------------------------------------------------
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, ]　　　　#  get\post都要加的裝飾器

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

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

 def dispatch_request(self): print('Index') return 'Index!'　　　　返回最終的字符串 函數處理完

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

 

四、請求相關

from flask import request

request.method
request.args        # url攜帶的數據
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/' + secure_filename(f.filename))
      

 

五、響應

響應相關信息
return "字符串"
return render_template('html模板路徑',**{})
return redirect('/index.html')
return jsonify({"a":1})
response = make_response(render_template('index.html'))

response是flask.wrappers.Response類型
response.delete_cookie('key')
response.set_cookie('key', 'value')
response.headers['X-Something'] = 'A value'
return response  

 

六、模板渲染

flask模板渲染使用的是jinja2，其語法和django無差異，但更接近python的語法

<h1>自定義函數</h1>
    {{ww()|safe}}        //並渲染標籤ww()返回'<h1>Wupeiqi</h1>' Markup,與django的mark_safe 同樣
　　from django.utils.safestring import marf_safe 
{{sb(1,2)}}  {{ 1|db(2,3)}}


{% extends "layout.html"%}　　　　# 繼承以後能夠經過 {{ super()}} 訪問父模板的資源
{% block content %}
　　{% include "form.html" %}

{% endblock %} 若果傳入的是一個函數須要()，與django不一樣，不會自動執行

<img src="/static/code.png">
<img src="{{ url_for('static',filename='code.png') }}">

 

jinja2 特有 宏

{% macro input(name, type='text', value='') %}
    <input type="{{ type }}" name="{{ name }}" value="{{ value }}">
{% endmacro %}
其中macro相似於def
 每 {{ input('n1') }}一次便會生成一次input標籤

{% import 'm.txt' as my%}


{{my.myMarco(xx)}}


在m.txt中聲明瞭另外一個宏

 

 

七、session

除請求對象以外，還有一個 session 對象。它容許你在不一樣請求間存儲特定用戶的信息。它是在 Cookies 的基礎上實現的，對 Cookies 進行密鑰簽名要使用會話，你須要設置一個密鑰。app.secret_key = ""

當請求到來時,flask讀取cookie中對應的值，將該值解密反序列化成字典，放入內存一邊視圖函數使用，當請求結束時，
會讀取內存中字典的值，進行序列化+加密寫到cookies

 

from flask import session
session["username"] = "xxx"
session.pop(key,defaut)

pip3 install Flask-Session
        
run.py
    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()

session.py
    #!/usr/bin/env python
    # -*- coding:utf-8 -*-
    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)

 

自定義session

 

‘session:隨機字符串1’:‘q23asifaksdfkajsdfasdf’

 

 

 

八、閃現

# 基於session的一次性存取機制， 利用了pop
from flask import Flask, flash, request, get_flashed_messages

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


@app.route('/set')
def index2():
    v = request.args.get('p')
    flash(v，"error")　　　　　　#  指定消息類型
    return 'ok'


@app.route('/')
def index1():
    messages = get_flashed_messages()　　　　# 還能夠經過category_filter= ["error",]　　來獲取存取的值
    print(messages)
    return "Index1"


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

 

九、中間件

app.run()
->
run_simple(host, port, self, **options)    # 943行，Flask實例，一旦請求到來調用self的call方法，（最先的一步）
->
def __call__(self, environ, start_response):
    return self.wsgi_app(environ, start_response)
在執行wsgi_app方法先後進行處理

class Middleware:
    def __init__(self, old_wsgi_app):
        self.old_wsgi_app = old_wsgi_app

    def __call__(self, *args, **kwargs):　　　　# environ,start_response請求相關的全部數據，用於設置響應相關數據　
        #  func1
        result = self.old_wsgi_app(*args, **kwargs)　　　　#  在werkzeug執行以前，更原始的數據
        # func2
        return result

if __name__ == '__main__':
    app.wsgi_app = Middleware(app.wsgi_app)　　# 將Middleware實例對象賦給app.wsgi，
　　　　　　　　　　　　　　　　　　->當wsgi_app調用時執行Middleware的call方法，裏面有原生的wsgi_app方法
    app.run()

 

十、藍圖（blueprint）

from flask import Blueprint 

目錄結構的劃分

 

十一、特殊的裝飾器

 

@app.before_first_request      程序開始的第一次請求到來前執行此函數

@app.before_request            相似於中間件的process_request，若是return直接走全部的after_request          　　  
    
@app.after_request             相似於中間件的process_response,對應的函數接收一個response參數，並將其返回

@app.errorhandler(404)
def page_not_found(error):
    return 'This page does not exist',404        # 捕捉404錯誤，返回處理

@app.template_global()    　　　　 
@app.template_filter()

 

擴展

RequestContext對象 封裝了app,request session三個屬性

 

 

只要重寫這個類就能夠存到redis裏

 

爲每一個線/協程建立一個獨立的空間，使得線/協程對本身的空間中的數據進行操做（數據隔離）。如同flask保證多個request,session存放於一塊，而沒有數據混淆
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()

 

 

 

 

from flask import Flask
from werkzeug.wsgi import DispatcherMiddleware
from werkzeug.serving import run_simple

app1 = Flask("app1")
app1.config['DB'] = 123

app2 = Flask("app2")
app1.config['DB'] = 456


@app1.route('/web')
def web():
    print('web')
    return '12213'


@app1.route('/news')
def news():
    print('news')
    return '12213'


@app2.route('/admin')
def admin():
    print('admin')
    return '12213'


@app2.route('/article')
def article():
    print('article')
    return '12213'


"""
/web
/new
/app2/admin
/app2/article
"""
app = DispatcherMiddleware(app1, {
    '/app2': app2,
})
if __name__ == '__main__':
    run_simple(hostname='127.0.0.1', port=5000, application=app)

mult_app

from multi_app import app1
from multi_app import app2

with app1.app_context():
    pass  # 爲app1建立數據庫
    with app2.app_context():
        pass  # 爲app2建立數據庫

離線腳本