flask

时间 2019-11-18

标签 flask 栏目 Python 繁體版

原文原文链接

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，sessionhtml

一、配置文件

flask的配置文件是以个flask.config.Config对象，继承了字典python

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默认配置

修改配置信息:mysql

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="')git

@app.route("/index/<int:post_id>")　　# 类型转换　　float、path、default、string、any、int、uuid、github

# 一旦有<>函数就必须接受这个关键字参数，另外还能够指定请求的参数 methods=["GET"，“POST”]　　web

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 = ""redis


当请求到来时,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

目录结构的划分sql

十一、特殊的装饰器

@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建立数据库

离线脚本