flask入门

Flask介绍

Flask是一个使用 Python 编写的轻量级 Web 应用框架。其 WSGI 工具箱采用 Werkzeug(Django使用的是wsgiref) ，模板引擎则使用 Jinja2 。

Flask特点:

• 短小精悍,可拓展强,第三方组件丰富

与Django的比较:

• 大而全，内部提供：ORM、Admin、中间件、Form、ModelForm、Session、缓存、信号、CSRF；

tornado:

• 短小精悍+异步非阻塞

简单的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)

Flask版Hello Word

from flask import Flask

app = Flask(__name__) # 一个Flask类的对象

@app.route('/index') # 此乃路由也
def index():
    return 'Hello World'

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

路由系统

路由与函数绑定的两种方式

装饰器方式:

from flask import Flask
app = Flask(__name__)

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

# 参数介绍 rule=匹配路由,methods=允许请求方式,endpoint=反向解析的名字

直接绑定

from flask import Flask
app = Flask(__name__)

def order():
    return 'Order'

app.add_url_rule('/order',view_func=order)
# route内部就是用了这种方式

动态路由匹配

from flask import Flask
app = Flask(__name__)

@app.route('/index/<int:nid>',methods=['GET','POST'])
def index(nid):
    return "Index"

# <类型:关键字>
# 类型转换器介绍
DEFAULT_CONVERTERS = {
    'default':          UnicodeConverter,
    'string':           UnicodeConverter,
    'any':              AnyConverter,
    'path':             PathConverter,
    'int':              IntegerConverter,
    'float':            FloatConverter,
    'uuid':             UUIDConverter,
}

反向解析

from flask import Flask,url_for
app = Flask(__name__)

@app.route('/index',methods=['GET','POST'],endpoint='n1')
def index():
    v1 = url_for('n1') # 关键字传参
    print(v1)
    return "Index"

　　解析的默认名字是函数名,当为视图函数使用装饰器时一定要修复,否则Flask项目会因存在因装饰器装饰多个视图函数,产生相同的endpoint而无法启动

自定义正则匹配

from flask import Flask, views, url_for
from werkzeug.routing import BaseConverter
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
# 添加到flask中
app = Flask(import_name=__name__)
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()

路由重定向

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

# 当访问此路由时会定位到/new
@app.route('/index',methods=['GET','POST'],redirect_to='/new')
def index():
    return "老功能"

@app.route('/new',methods=['GET','POST'])
def new():
    return '新功能'


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

子域名获取

from flask import Flask,render_template,redirect
app = Flask(__name__)
app.config['SERVER_NAME'] = 'oldboy.com:5000'


@app.route("/dynamic", subdomain="<username>")
def xxxxxxx(username):
    print(username)
    return 'xxxxx'

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

路由的一些参数

@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最后的 / 符号是否严格要求，
subdomain=None,             子域名访问

视图

FBV的两种形式

# 方式一
from flask import Flask,render_template,redirect,views
app = Flask(__name__)
class IndexView(views.View):
    methods = ['GET']  # 允许方法
    decorators = [wapper, ] # 装饰器

    def dispatch_request(self):
        print('Index')
        return 'Index!'

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


# 方式二
class IndexView(views.MethodView):
    methods = ['GET']
    decorators = [wapper, ]

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

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

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

请求和响应

# 请求数据获取
from flask import request
# request.method
# request.args
# request.form
# 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
# request.get_data() 请求元数据
# obj = request.files['the_file_name']
# obj.save('/var/www/uploads/' + secure_filename(f.filename))

# 返回响应
# return "字符串"
from flask import render_template
# return render_template('html模板路径',**{})
from flask import redirect
# return redirect('/index.html') 重定向

# 设置响应头和cookie
from flask import make_response
# 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

有坑

读取文件form表单应设置enctype="multipart/form-data"

模板

Markup等价django的mark_safe

Session

除请求对象之外，还有一个 session 对象。它允许你在不同请求间存储特定用户的信息。它是在 Cookies 的基础上实现的，并且对 Cookies 进行密钥签名要使用会话，你需要设置一个密钥。

• 设置：session['username'] ＝ 'xxx'

• 删除：session.pop('username', None)

from flask import Flask,render_template,request,redirect,session

app = Flask(__name__) # 一个Flask类的对象
app.secret_key = 'u2jksidjflsduwerjl'  # 秘钥
app.debug = True

@app.route('/login',methods=['GET',"POST"])
def login():
    if request.method == 'GET':
        return render_template('login.html')
    user = request.form.get('user') # 获取POST传过来的值
    pwd = request.form.get('pwd') # 获取POST传过来的值
    if user == 'alex' and pwd == '123':
        # 用户信息放入session
        session['user_info'] = user
        return redirect('/index')
    else:
        return render_template('login.html',msg ='用户名或密码错误')
        # return render_template('login.html',**{'msg':'用户名或密码错误'})


@app.route('/logout')
def logout():
    session.pop('user_info')
    return redirect('/login')

配置文件

flask中的配置文件是一个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),
'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,
    }

设置方式

# 方式一：
app.config['DEBUG'] = True
PS： 由于Config对象本质上是字典，所以还可以使用app.config.update(...)
# 方式二:
app.config.from_object("python类或类的路径")
# 这样写类,
class BaseConfig(object):
    DEBUG = True
    SECRET_KEY = "asudflkjdfadjfakdf"

　　

特殊装饰器

from flask import Flask, Request, render_template

app = Flask(__name__, template_folder='templates')
app.debug = True

# 只有第一次请求会执行
@app.before_first_request
def before_first_request1():
    print('before_first_request1')

# 视图函数之前
@app.before_request
def before_request1():
    Request.nnn = 123
    print('before_request1')

# 视图函数之后
@app.after_request
def after_request1(response):
    print('before_request1', response)
    return response

# 自定义错误信息
@app.errorhandler(404)
def page_not_found(error):
    return 'This page does not exist', 404

# 全局过滤函数
@app.template_global()
def sb(a1, a2):
    return a1 + a2

# 可以向模板上下文中自动注入变量或函数(必须是字典)，在模板中调用
@app.context_processor 
def inject_user():
      return dict(user=g.user)

# 模板的过滤函数 {{a1|db(a2,a3)}}
@app.template_filter()
def db(a1, a2, a3):
    return a1 + a2 + a3


@app.route('/')
def hello_world():
    return render_template('hello.html')


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

　　before_request的执行顺序是按代码顺序执行的，after_request的执行数顺序是反向执行的,当before_request返回内容的时候,after_request会全部执行,之前版本的Django也是这样做的

闪现威慑,点燃警告

message是一个基于Session实现的用于保存数据的集合，其特点是：使用一次就删除。

from flask import Flask,session,flash,get_flashed_messages
app = Flask(__name__)
app.secret_key = 'asdfasdfasdf'

@app.route('/x1',methods=['GET','POST'])
def login():
    flash('存一个',category='x1') # category是作为关键字,可以不设置
    return "视图函数x1"

@app.route('/x2',methods=['GET','POST'])
def index():
    data = get_flashed_messages(category_filter=['x1']) # category_filter根据关键字取,可以不设置
    return "视图函数x2"

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

中间件

falsk中间件利用了装饰器的性质,他执行的时request,session还没有生成

from flask import Flask
app = Flask(__name__)
app.secret_key = 'asdfasdfasdf'

@app.route('/x2',methods=['GET','POST'])
def index():
    return "x2"


class Middleware(object):
    def __init__(self,old_wsgi_app):
        """
        服务端启动时，自动执行
        :param old_wsgi_app:
        """
        self.old_wsgi_app =old_wsgi_app

    def __call__(self, environ, start_response):
        """
        每次有用户请求道来时
        :param args:
        :param kwargs:
        :return:
        """
        print('before')
        from flask import session,request
        obj = self.old_wsgi_app(environ, start_response)
        print('after')
        return obj

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

蓝图

Flask 用 蓝图（blueprints） 的概念来在一个应用中或跨应用制作应用组件和支持通用的模式。

# 简单的蓝图项目结构
| - projectName  
    | - app  //程序包  
        | - templates //jinjia2模板  
        |- static //css,js 图片等静态文件  
        | - main  //py程序包 ，可以有多个这种包，每个对应不同的功能  
            | - __init__.py  
            |- errors.py  
            |- forms.py  
            |- views.py  
        |- __init__.py  
        |- email.py //邮件处理程序  
        |- models.py //数据库模型  
    |- requirements.txt //列出了所有依赖包以及版本号，方便在其他位置生成相同的虚拟环境以及依赖  
    |- config.py //全局配置文件，配置全局变量  
    |- manage.py //启动程序

在app中的py文件中创建一个蓝图

from flask import Blueprint

us = Blueprint('us',__name__)

@us.route('/info')
def info():
    return 'info'

注册蓝图

创建flask的唯一实例,并且在初始化的时候注册蓝图

在app中的init完成创建实例与注册蓝图的过程

from flask import Flask
from .app import views

app = Flask(__name__)
app.register_blueprint(views.us)

项目的启动文件

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

蓝图的几个作用

1. 为一个蓝图下的路由设置before_request和after_request
2. 蓝图URL前缀：xxx = Blueprint('account', __name__,url_prefix='/xxx')
3. 蓝图子域名：xxx = Blueprint('account', __name__,subdomain='admin')

# 前提需要给配置SERVER_NAME： app.config['SERVER_NAME'] = 'wupeiqi.com:5000'
# 访问时：admin.wupeiqi.com:5000/login.html

信号

Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为。

pip3 install blinker

内置信号及执行顺序

flask.signals.py
appcontext_pushed = _signals.signal('appcontext-pushed') #  请求上下文push时执行
request_started = _signals.signal('request-started') #  请求到来前执行
 
如果有render:
	before_render_template = _signals.signal('before-render-template') # 模板渲染前执行
	template_rendered = _signals.signal('template-rendered') # 模板渲染后执行

request_finished = _signals.signal('request-finished') # 请求结束后执行
如果视图函数有异常：
	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_popped = _signals.signal('appcontext-popped') # 请求上下文pop时执行


如果使用信号：
	message_flashed = _signals.signal('message-flashed') # 调用flask在其中添加数据时，自动触发

信号的使用:

from flask import signals
def func(*args,**kwargs):
    pass
#信号中注册函数
signals.xxx.connect(func)

自定义信号

from flask import Flask, flash
from flask.signals import _signals

app = Flask(__name__)

xinhao = _signals.signal("xinhao")  # 创建信号


# 定义函数
def func1(*args, **kwargs):
    print("func1", args, kwargs)


def func2(*args, **kwargs):
    print("func2", args, kwargs)


# 将函数注册到信号中,添加到这个列表
xinhao.connect(func1)
xinhao.connect(func2)


@app.route("/zzz")
def zzz():
    xinhao.send(sender='xxx', a1=123, a2=456)  # 触发这个信号，执行注册到列表中的所有函数,这里的参数个上面函数的参数一致
    return "发送信号成功"


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

# 打印结果
# func1 (None,) {'sender': 'xxx', 'a1': 123, 'a2': 456}
# func2 (None,) {'sender': 'xxx', 'a1': 123, 'a2': 456}

问题：信号和before_request区别？

• before_request，可以控制请求是否可以继续往后执行。
• 信号，在原来的基础增加额外的操作和值。

多app应用

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

app01 = Flask('app01')
app02 = Flask('app02')

@app01.route('/login')
def login():
    return 'app01.login'

@app02.route('/index')
def index():
    return 'app02.index'


dm = DispatcherMiddleware(app01,{
				'/app02':        app02,
			})

if __name__ == '__main__':
	run_simple('localhost', 5000,dm)

源码:

class DispatcherMiddleware(object):

    """Allows one to mount middlewares or applications in a WSGI application.
    This is useful if you want to combine multiple WSGI applications::

        app = DispatcherMiddleware(app, {
            '/app2':        app2,
            '/app3':        app3
        })
    """

    def __init__(self, app, mounts=None):
        self.app = app
        self.mounts = mounts or {}
    # 请求来临时__call__方法被执行
    def __call__(self, environ, start_response):
        script = environ.get('PATH_INFO', '')
        path_info = ''
        while '/' in script:
            if script in self.mounts:  # 如果路由存在与self.mounts中,就说明是其他app的
                app = self.mounts[script]
                break
            script, last_item = script.rsplit('/', 1) # 倒序分割
            path_info = '/%s%s' % (last_item, path_info)# 将分割出去的在拼接起来
        else:
            app = self.mounts.get(script, self.app)
        original_script_name = environ.get('SCRIPT_NAME', '')
        environ['SCRIPT_NAME'] = original_script_name + script
        environ['PATH_INFO'] = path_info
        return app(environ, start_response) # app()执行

离线脚本编写

falsk中数据是在请求来临时通过RequestContext存在Local对象中,但是如果不是通过请求的方式我们要怎样获取数据

with falskobj.app_context():
    pass

源码:

class Flask(_PackageBoundObject):
    def app_context(self):
        return AppContext(self)

with 语法会去执行后面语句返回对象的__enter__方法,with语句结束后会执行__exit__方法
class AppContext(object):
    def __enter__(self):
        self.push()
        return self

    def __exit__(self, exc_type, exc_value, tb):
        self.pop(exc_value)

        if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None:
            reraise(exc_type, exc_value, tb)

这也是为什么使用LocalStack对Local对象进行操作？

目的是想要将local中的值维护成一个栈，例如：在多app应用中编写离线脚本时，可以实用上。
from m_app import app01,app02
from flask import current_app
"""
{
	1231: {
		stack: [app01,app02,]
		}
}
"""

with app01.app_context():
    print(current_app)
     with app02.app_context():
          print(current_app)
    print(current_app)