Flask基础

flask是什么？

它是一个Python编写微web框架，包括了一个核心两个库（jinja2模板引擎和WSGI工具集）

创建一个独立的虚拟环境

deactivate 退出

mkvirtualenv flask 创建虚拟环境

workon flask 切换到创建的虚拟环境

安装flask

pip install flask

包名及版本	功能
Jinja2-2.10	渲染模板引擎工具集
MarkupSafe-1.1.0	可以识别HTML转义规则。HTML字符转义工具集
Werkzeug-0.14.1	Web 框架的底层库，提供了请求及响应以及开发服务器的功能，简称WSGI工具集
click-7.0	命令行工具集
itsdangerous-1.1.0	加密工具集

我们安装这5个包，这就是一个基本的flask

✔提示：这些库均由Flask团队开发

创建一个简单的flask

from  flask import Flask

app=Flask(__name__)

@app.route("/")
def index():
    return "ojbk"


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

源码分析

启动

启动代码为app.run()

    def __call__(self, environ, start_response):
        return self.wsgi_app(environ, start_response)

核心

    def wsgi_app(self, environ, start_response):
        ctx = self.request_context(environ)
        error = None
        try:
            try:
                ctx.push()
                response = self.full_dispatch_request()
            except Exception as e:
                error = e
                response = self.handle_exception(e)
            except:  # noqa: B001
                error = sys.exc_info()[1]
                raise
            return response(environ, start_response)
        finally:
            if self.should_ignore_error(error):
                error = None
            ctx.auto_pop(error)

run

run主要做的就是用werkzeug来启动服务run_simple

    def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):       
        if os.environ.get("FLASK_RUN_FROM_CLI") == "true":
            from .debughelpers import explain_ignored_app_run

            explain_ignored_app_run()
            return

        if get_load_dotenv(load_dotenv):
            cli.load_dotenv()

            # if set, let env vars override previous values
            if "FLASK_ENV" in os.environ:
                self.env = get_env()
                self.debug = get_debug_flag()
            elif "FLASK_DEBUG" in os.environ:
                self.debug = get_debug_flag()

        # debug passed to method overrides all other sources
        if debug is not None:
            self.debug = bool(debug)

        _host = "127.0.0.1"
        _port = 5000
        server_name = self.config.get("SERVER_NAME")
        sn_host, sn_port = None, None

        if server_name:
            sn_host, _, sn_port = server_name.partition(":")

        host = host or sn_host or _host
        # pick the first value that's not None (0 is allowed)
        port = int(next((p for p in (port, sn_port) if p is not None), _port))

        options.setdefault("use_reloader", self.debug)
        options.setdefault("use_debugger", self.debug)
        options.setdefault("threaded", True)

        cli.show_server_banner(self.env, self.debug, self.name, False)

        from werkzeug.serving import run_simple

        try:
            run_simple(host, port, self, **options)
        finally:
            # reset the first request information if the development server
            # reset normally.  This makes it possible to restart the server
            # without reloader and that stuff from an interactive shell.
            self._got_first_request = False

Flask基本响应

from flask import Flask,render_template,redirect,jsonify,request
app=Flask(__name__)

'''
1 返回字符串
2 返回模板：render_template
3 跳转：redirect
4 json返回

'''

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

    return "zx"

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

@app.route("/home")
def home():
    return redirect("/index")

@app.route("/json")
def json1():
    data_dic={'name':"zx",'age':18}
    return jsonify(data_dic)


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

基本配置

可配置的选项

    default_config = ImmutableDict(
        {
            "ENV": None,
            "DEBUG": None,
            "TESTING": False,
            "PROPAGATE_EXCEPTIONS": None,
            "PRESERVE_CONTEXT_ON_EXCEPTION": None,
            "SECRET_KEY": None,
            "PERMANENT_SESSION_LIFETIME": timedelta(days=31),
            "USE_X_SENDFILE": False,
            "SERVER_NAME": None,
            "APPLICATION_ROOT": "/",
            "SESSION_COOKIE_NAME": "session",
            "SESSION_COOKIE_DOMAIN": None,
            "SESSION_COOKIE_PATH": None,
            "SESSION_COOKIE_HTTPONLY": True,
            "SESSION_COOKIE_SECURE": False,
            "SESSION_COOKIE_SAMESITE": None,
            "SESSION_REFRESH_EACH_REQUEST": True,
            "MAX_CONTENT_LENGTH": None,
            "SEND_FILE_MAX_AGE_DEFAULT": timedelta(hours=12),
            "TRAP_BAD_REQUEST_ERRORS": None,
            "TRAP_HTTP_EXCEPTIONS": False,
            "EXPLAIN_TEMPLATE_LOADING": False,
            "PREFERRED_URL_SCHEME": "http",
            "JSON_AS_ASCII": True,
            "JSON_SORT_KEYS": True,
            "JSONIFY_PRETTYPRINT_REGULAR": False,
            "JSONIFY_MIMETYPE": "application/json",
            "TEMPLATES_AUTO_RELOAD": None,
            "MAX_COOKIE_SIZE": 4093,
        }
    )

配置方式

from  flask import Flask
app = Flask(__name__)
#方式一
# 只能配置下面俩个
# app.debug=True
# app.secret_key="asdjja"
#方式二，以字典的形式
# app.config['DEBUG']=True
#方式三 ，以文件的形式
# app.config.from_pyfile("settings.py")
#方式四 ，以类的形式
app.config.from_object("settings.test")


@app.route("/")
def index():
    return "ok"

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

类的形式

一般都是以类的形式，可以通过继承的方式实现公共提取，部分修改，行程多套配置

class common:
	#公共配置
    
#开发配置
class test(common):
    DEBUG = True

#生产配置
class onlin(common):
    DEBUG = False

路由本质

#我们的路由是依靠这个实现的
@app.route("/")

#源码
    def route(self, rule, **options):
        def decorator(f):
            endpoint = options.pop("endpoint", None)
            self.add_url_rule(rule, endpoint, f, **options)
            return f

        return decorator

当代码加载，就会直接运行上面的函数，得到运行结果如下

#实际上就是一个装饰器
@decorator

所以在view里面，装饰器路由就等同于

    @setupmethod
    def add_url_rule(
        self,
        rule,
        endpoint=None,
        view_func=None,
        provide_automatic_options=None,
        **options
    ):

app.add_url_rule(rule, endpoint, f, **options)

最终

就和Django的路由映射非常的相似了

#路由和函数关系映射
app.add_url_rule("/index", view_func=index)

CBV

#简单模式
#默认执行的就是dispatch_request方法
#name="index1"用于方法和路由的反射

 def login(f):
 	pass
 class IndexView(views.View):
 	 #允许的请求方式
     methods = ['GET']
     #给dispatch_request加装饰器
     #decorators = [login, ]
     def dispatch_request(self):
         print('Index')
         return 'Index!'

app.add_url_rule('/index1', view_func=IndexView.as_view(name='index1'))

#通常用此方式
#会根据请求方式直接找对应的方法
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

路由参数

@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"]

redirect_to

这个重定向是不执行里面代码直接重定向的，和之前的不同

@app.route("/index/",endpoint="a",methods=["POST","GET"],strict_slashes=True,redirect_to="/index2")
def index():
    print(url_for("a"))
    return "ok"

@app.route("/index2")
def index2():
    return "吃饭去了"

路由转化器

本质

就是通过路由传递参数

from  flask import Flask
app = Flask(__name__)

@app.route("/<string:flag>")
def index(flag):
    return f"jason is sb ? {flag}"

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

自定义路由转化器

#1 写类，继承BaseConverter
#2 注册：app.url_map.converters['regex'] = RegexConverter
# 3 使用：@app.route('/index/<regex("d+"):nid>')  正则表达式会当作第二个参数传递到类中
from flask import Flask, views, url_for
from werkzeug.routing import BaseConverter

app = Flask(import_name=__name__)

class RegexConverter(BaseConverter):
    """
    自定义URL匹配正则表达式
    """
    #__init__固定写法
    def __init__(self, map, regex):
        super(RegexConverter, self).__init__(map)
        self.regex = regex

    def to_python(self, value):
        """
        路由匹配时，匹配成功后传递给视图函数中参数的值
        """
        return int(value)+123

    def to_url(self, value):
        """
        使用url_for反向生成URL时，传递的参数经过该方法处理，返回的值用于生成URL中的参数
        """
        #固定写法，调用父类的
        val = super(RegexConverter, self).to_url(value)
        return val+"json"

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

#nid会先去to_python进行处理完，之后再把结果返回给函数的nid
@app.route('/index/<regex("d+"):nid>')
def index(nid):
   # print(nid)
    print(url_for('index', nid='888'))
    return 'Index'

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

常见模板语法

数据

USERS = {
    1:{'name':'张三','age':18,'gender':'男','text':"道路千万条"},
    2:{'name':'李四','age':28,'gender':'男','text':"安全第一条"},
    3:{'name':'王五','age':18,'gender':'女','text':"行车不规范"},
}


def func1(flag):
    return Markup(f"<h1>jason dsb ?{flag}</h1>" )

@app.route("/index")
def index():
    data=USERS
    flag=True
    htm="<h1>jason dsb</h1>"
    return render_template("index1.html",data=data,flag=flag,name="sb",name1="sb1",htm=htm,func=func1)

模板

//循环
{% for k,v in data.items() %}
        <tr>
            <td>{{k}}</td>
            <td>{{v.name}}</td>
            <td>{{v['name']}}</td>
            <td>{{v.get('name')}}</td>
            <td><a href="/detail/{{k}}">查看详细</a></td>
        </tr>
{% endfor %}

//条件判断
{% if flag %}
    {{name}}
{% else %}
    {{name1}}
{% endif %}
 <br>

//两种后端直接传递html
{{htm|safe}}

{{func("yes")}}

session

原理

我们知道flask默认是没有连接数据库的，那么它的session是怎么存储的呢？

其实和token的原理类似，将数据存储在客户端的cookies里面，然后到后端解密取值

from  flask import Flask,session
app = Flask(__name__)
#加密的盐
app.secret_key="zxxzxzcas"
#存在客户端的cookies名
app.config["SESSION_COOKIE_NAME"]="zx"

# app.session_interface

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

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

#客户端存的cookies就类似这个，后面的是加密的数据
zx:sadasduasgdhjasghjdjasgdhja

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

request&response高级

注意

rerequest不是通过传参的方式拿到的，而是import

response想要设置cookie或者响应头，需要先创建respon对象

可以是返回字符串和静态界面

response = make_response("内容")

from flask import Flask,request,make_response,render_template

app = Flask(__name__)

@app.route('/login.html', methods=['GET', "POST"])
def login():
    # 请求相关信息
    # request.method  提交的方法
    # request.args  get请求提及的数据
    # request.form   post请求提交的数据
    # request.values  post和get提交的数据总和
    # request.cookies  客户端所带的cookie
    # request.headers  请求头
    # request.path     不带域名，请求路径
    # request.full_path  不带域名，带参数的请求路径
    # request.script_root
    # request.url           带域名带参数的请求路径
    # request.base_url      带域名请求路径
    # request.url_root      域名
    # request.host_url      域名
    # request.host          127.0.0.1:500
    # 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({'k1':'v1'})

    # response = make_response(render_template('index.html'))

    # response是flask.wrappers.Response类型
    #删除cookie
    # response.delete_cookie('key')
    #设置cookie
    # response.set_cookie('key', 'value')
    #设置响应头
    # response.headers['X-Something'] = 'A value'
    # return response

    #返回字符串
    #response = make_response("内容")
    return response

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

local原理

我们知道如果是使用Django的话，每一次的request都是直接写写在方法参数的，这样就很好理解，一次请求就会把请求的参数封装为一个request

flask

但是flask我们使用session，竟然是import，是不是不理解，其实它底层是这样实现的

_request_ctx_stack是一个LocalStack实例的全局变量，来一个请求就把ctx存到LOCAL对象的storage中去，以线程id，协程id为区分

ctx.push()里面执行了
_request_ctx_stack.push(self)：self是ctx
                
LocalStack里面的push方法,obj参数是ctx
def push(self, obj):
    rv = getattr(self._local, "stack", None)
    if rv is None:
       self._local.stack = rv = []
       #把obj,也就是ctx存到了LOCAL对象中的storage,存的格式为： storage={"执行id":{'stack':[ctx,]}}
       
    rv.append(obj)
    return rv
#rv是啥     
rv = getattr(self._local, "stack", None)这个self._local是LOCAL对象

核心

有使用到getattr和setattr

    __local=LOCAL()
    __local.stack=[ctx,]
    storage={stack:[ctx,]}
    storage={"执行id":{'stack':[ctx,]}}

    print(__local.stack)
    storage["执行id"].get("stack")[-1]

    storage["执行id"]['stack']

闪现

注意

1.由于闪现是基于session的所以必须要先设置

app.secret_key = 'asfkjabm'

2.闪现的k值可以重复，打印出来是元组的形式展示

原理**

flash是基于session实现的

特点

闪现的取值只能出现在一次请求内部，它会先去请求的内部变量去找值，没有的话，去session，pop获取存在内部变量里面，这个变量的生命周期就是一次请求内

然后第二次请求原先的session内的值已经pop掉了，内存中也清空了

使用

#设置闪现值
@app.route('/index')
def index():
	#默认闪现的，k值为message
    flash('超时错误')
    #也可以手动设置闪现值
    flash('哈哈',category="zx")
    return "ssdsdsdfsd"
    
#取闪现值
@app.route('/error')
def error():
	#with_categories默认为false，是否显示k值
	#category_filter默认为空，过滤器，显示选定的k的闪现值
    data = get_flashed_messages(with_categories=True,category_filter=('zx',))
    data2 = get_flashed_messages()
    print(data)
    print(data2)
    return  "ojbk"

请求扩展

说明

和Django的中间件很相似

源码

full_dispatch_request()

    def wsgi_app(self, environ, start_response):
        ctx = self.request_context(environ)
        error = None
        try:
            try:
                ctx.push()
                response = self.full_dispatch_request()
            except Exception as e:
                error = e
                response = self.handle_exception(e)
            except:  # noqa: B001
                error = sys.exc_info()[1]
                raise
            return response(environ, start_response)
        finally:
            if self.should_ignore_error(error):
                error = None
            ctx.auto_pop(error)

full_dispatch_request()

请求和响应部分的源码

    def full_dispatch_request(self):
        self.try_trigger_before_first_request_functions()
        try:
            request_started.send(self)
            rv = self.preprocess_request()
            #返回为None就执行视图函数
            if rv is None:
                rv = self.dispatch_request()
        except Exception as e:
            rv = self.handle_user_exception(e)
        return self.finalize_request(rv)

请求之前执行-before_request

按顺序执行，如果中途返回值不为None，直接断掉，去执行所有的响应过滤函数

@app.before_request
def sb():
    print(request)
    print("我是请求之前")

@app.before_request
def  sb1():
    print("我是请求之前2")
    #return "我是请求之前2的返回"

响应之后执行-after_request

执行顺序相反

@app.after_request
def process_response1(response):
    print('process_response1 走了')
    return response

@app.after_request
def process_response2(response):
    print('process_response2 走了')
    return response

第一次请求执行-before_first_request

原理就是定义个全部变量，第一次来赋值

 @app.before_first_request
 def first():
    print("我的第一次")

异常-teardown_request

无论有没有异常都会执行，会在视图函数执行后执行

@app.teardown_request
def ter(e):
    print(e)
    print("我是异常")

异常捕获-errorhandler

#要捕获的错误码
@app.errorhandler(500)
def error_404(arg):
    print(arg)
    #返回给前端，可以为界面
    return "500错误了"

全局定义函数-template_global

使用的时候不需要传递，直接在模板界面使用

@app.template_global()
def sb(a1, a2):
    return a1 + a2

{{sb(1,2)}}

全局过滤器-template_filter

@app.template_filter()
def db(a1, a2, a3):
    return a1 + a2 + a3
    
{{12|db(1,1)}}

中间件

重写源码，然后调用源码核心部分就行

from flask import Flask,flash,get_flashed_messages,request

app = Flask(__name__)

class MyMiddleware:
    def __init__(self,wsgi_app):
        self.wsgi_app=wsgi_app

    def __call__(self, environ, start_response):

        print("中间件前")
        res=self.wsgi_app(environ, start_response)
        print("中间件后")
        print(res)
        return res

@app.route('/index')
def index():
    return "zx"

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

偏函数

偏函数和闭包函数类似

from functools import partial
def test(a,b,c,d):
    return a+b+c+d

tes=partial(test,a=1,b=2)

print(tes(c=3,d=4))

参考链接

https://www.cnblogs.com/xiaoyuanqujing/category/1561311.html