Flask备注二（Configurations, Signals)

时间：2023-03-09 02:54:54
Flask备注二（Configurations, Signals)

Flask备注二（Configuration, Signals)

Flask是一个使用python开发Web程序的框架。依赖于Werkzeug提供完整的WSGI支持，以及Jinja2提供templates支持。Flask的设计理念是提供Micro以及方便的框架。"Micro"是因为除了提供基本特性功能的实现外，其他的功能（例如数据库访问）都是通过extension来完成。方便的特点是因为提供了简单易用的必要特性和功能，提供的功能包含：

  1. 内置的开发服务器和调试工具。
  2. 集成单元测试支持。
  3. 支持Templates（依赖JinJa2提供）
  4. 完整的WSGI支持（依赖Werkzeug提供）
  5. 安全Cookie和Sessions
  6. Thread-Locals，例如在每个Request里面的对象只在线程里面有意义。

Configuration

Flask对象中包含一个config属性，是一个包含所有的参数值的对象。Flask程序的参数值需要在启动时加载到程序中，基于此目的Flask支持在代码中或者配置文件中设置参数值。代码配置参数通过对config属性的操作完成，config对象本身是一个字典的子类，支持字典操作。例如：app.config['DEBUG'] = True 或者通过update方法

app.config.update(

    DEBUG = True,

    SECERET_KEY = ...

    )

有些特定的参数作为flask对象的属性，可以通过属性直接操作，例如：app.debug = True修改了'DEBUG'参数的值。

可配置参数列表

__Configuration__ __Description__
DEBUG 是否开启调试模式
TESTING 是否开启测试模式
PROGATION_EXCEPTIONS 是否开启EXCEPTION的传递，在DEBUG/TESTING模式下一定为TRUE。
PRESERVE_CONTEXT_ON_EXCEPTION 是DEBUG模式下，EXCEPTION是会保留request context方便进行数据监测。通过这个参数可以开启保留在非DEBUG模式，也可以去掉保留在DEBUG模式。
SECRECT_KEY 秘钥
SESSION_COOKIE_NAME 会话cookie名称
SESSION_COOKIE_DOMAIN 会话cookie的作用域，如果没有设置，服务器下的所有区域都可以用。(*TODO:domain & subdomains?*)
SESSION_COOKIE_HTTPONLY 会话cookie的httponly标签，默认为True。
SESSION_COOKIE_SECURE 会话cookie的secure标签，默认为True。
PERMANENT_SESSION_LIFETIME permanent session的生命周期，参数值为代表秒数的数值。
USE_X_SENDFILE 是否使用x-sendfile
LOGGER_NAME 日志名称
SERVER_NAME 服务器名称和端口(*TODO:domain & subdomains?*)
APPLICATION_ROOT 如果程序并不占据一个完整的domain或者subdomain，设置程序的入口路径。(*TODO:domain & subdomains?*)
MAX_CONTENT_LENGTH 最大可以接受的request的content length，超出则返回413。
SEND_FILE_MAX_AGE_DEFAULT 传递文件的默认最大时间，单位是秒。默认值是43200(12小时)。
TRAP_HTTP_EXCEPTIONS 是否将HTTP错误不返回而是当做普通错误对待，进入错误堆栈。需要调试Http错误信息来源可以使用。
TRAP_BAD_REQUEST_ERRORS 是否将BadRequest错误当做普通错误对待，同上。
PREFRRED_URL_SCHEME 生成URL时所用的scheme默认是http。
JSON_AS_ASCII 默认情况下JSON序列化为ascii码。设置为false则序列化为unicode并且jsonfiy自动以utf-8传播数据。
JSON_SORT_KEYS 默认情况下JSON序列化按照Key排序的，这样是为了防止包含无用的一些Http冗余信息。设为false，则不排序（不推荐）。
JSONIFY_PRETTYPRINT_REGULAR 默认为True，即Jsonify的返回是进行过排版优化的。

Configuration 切换

除了在代码中更改参数配置之外，也可以通过单独的文件进行参数配置。这样就可以根据发布环境的不同选择不同的配置文件，不需要修改代码而切换不同的配置版本。简单示例如下：

app = Flask(__name__)

app.config.from_object('yourapplication.default_settings')

app.config.from_envvar('YOURAPPLICATION_SETTINGS')

在上述示例中，通过fromobject函数从配置文件中读取所有的默认参数值，然后通过fromenvvar从环境变量YOURAPPLICATION_SETTINGS指向的文件中获取根据需求而改变的参数值。例如在开发调试版本中export YOURAPPLICATION_SETTINGS=/path/to/debug.cfg，在产品发行版本中export YOURAPPLICATION_SETTINGS=/path/to/release.cfg，这样就可以进行不同版本的配置参数的切换。

请确保参数配置文件被导入的比较早，因为很多扩展会需要参数配置的值。

除了通过文件进行参数配置可以进行参数版本切换之外，在代码中配置参数，也可以通过面向对象的继承和多态的特点完成参数版本切换。定义一个默认config对象，通过继承形成不同版本的config对象，在from_object中导入合适的对象来完成参数版本的切换。这种方式同样简单，只需要修改一句代码，但是依然需要修改代码。这种实现的示例如下：

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.config.from_object('configmodule.ProductionConfig')

Signals

Flask中提供低耦合的Signals进行程序模块间的信息传递。Flask基于blinker库实现signals的机制，自身框架提供了一些常用信号，另外一些扩展也提供其它的信号。在功能上来看，signals和flask程序中的一些handler(decorated function)类似，但是两者的设计和实现原则有着很大的不同。

  • signal是异步的操作，进行到预定流程时，Signal会触发所有的subscriber，然后继续自身流程。subscriber函数不能改变和终止Signal所在流程。
  • handler是通过函数的修改，将函数的内容加入到handler所在的流程中。因此在这个函数中可以改变或者终止hanler所在流程。
  • signal通过connect函数完成订阅，通过disconnect函数解除订阅。而handler是所在模块在global范围执行的代码替换，在import时执行，无法反向操作。

由于signals的特点，我们可以临时的订阅signal来检测应用程序在获取请求后的状态，通过这种方式可以将singal很好的使用在单元测试里面。下述是一个使用context manager以及generator来完成临时订阅的示例：

from flask import template_rendered

from contextlib import contextmanager

@contextmanager

def captured_templates(app):

    recorded = []

    def record(sender, template, context, **extra):

        recorded.append((template, context))

        template_rendered.connect(record, app)

        try:

            yield recorded

        finally:

            template_rendered.disconnect(record, app)

这样我们可以在下述单元测试中使用此临时订阅，验证程序正常运行状态:

with captured_templates(app) as templates:

    with app.test_client() as c:

        rv = c.get('/')

        assert rv.status_code == 200

        assert len(templates) == 1

        template, context = templates[0]

        assert template.name == 'index.html'

        assert len(context['items']) == 10

注意：临时订阅函数中包含**extra参数，这样可以确保在singal包含其他参数时，订阅也可以触发成功。

同样的我们可以使用signal的connect_to函数来完成临时订阅，这个订阅本身包含contextmanager，可以自动完成释放订阅。因此更简便。示例如下：

from flask import template_rendered

def captured_templates(app, recorded, **extra):

    def record(sender, template, context):

        recorded.append((template, context))

    return template_rendered.connected_to(record, app)

templates = []

with captured_templates(app, templates, **extra):

    ...

    template, context = templates[0]

另外一种signal的订阅方式类似于handler，通过decorator完成。更简单但是失去了随时解除订阅的特点，根据情况使用。

from flask import template_rendered

@template_rendered.connnect_via(app)

def when_template_rendered(sender, template, context, **etra):

    print 'Tempalte %s is rendered with %s' % (template.name, context)

自定义Signal

通过blinker库，在程序中可以创建自定义signal，

  1. 首先创建一个自定义的signal命名空间。

    from blinker import Namespace
    
xx_signals = Namespace()

  2. 然后通过这命名空间创建自定义signal。其中'xxx-done'为信号名称。

    xxx_done = xx_signals.signal('xxx-done')

  3. 通过send函数触发所有的订阅者，send函数的第一个参数sender，可以是类，也可以是当前的app。send函数的其他参数包含需要传递给订阅者的必要信息。

    class Xxx(object):
    
    ...
    def done(self):
    
    xxx_done.send(self)

如果需要传递当前app给send函数，需要使用currentapp.getcurrentobject()获取当前app的对象。不要直接使用current_app，因为这只是一个符号代理。

基本信号列表

Flask提供的信号列表如下：

  • flask.template_rendered

    这个信号在template被成功加载时触发，signal发送的信息包含template: 加载的模板示例；context：上下文环境字典。使用示例：

    def log_template_renders(sender, template, context, **extra):
    
    sender.logger.debug('Rendering template "%s" with context %s',
    
                        template.name or 'string template',
    
                        context)

from flask import template_rendered
    
    template_rendered.connect(log_template_renders, app)

  • flask.request_started

    这个信号在flask针对request做任何处理之前，但是已经创建了request的context时触发。因此此时被触发的subscriber可以使用request的上下文环境以及全局代理。使用示例：

    def log_request_started(sender, **extra):
    
    sender.logger.debug('Request is here and request context is set up')

from flask import request_started
    
request_started.connect(log_request_started, app)

  • flask.request_finished

    这个信号在reponse返回之前触发，signal包含reponse:返回的reponse。使用实例：

    def log_request_finished(sender, response, **extra):
    
    sender.logger.debug('Requset context is about to close down...
    
                        Response: %s', response)

from flask import request_finished
    
request_finished.connect(log_request_finished, app)

  • flask.gotrequestexception

    这个信号在request处理过程中发生异常时触发，信号的触发在异常的标准处理之前。signal发送的信息包含exception:异常信息。使用示例：

    def log_request_exception(sender, exception, **extra):
    
    sender.logger.debug('Got exception during processing request, Exception: %s', exception)

from flask import got_request_exception
    
got_request_exception.connect(log_request_exception, app)

  • flask.requesttearingdown

    这个信号在request销毁时触发，信号的subscriber在teardown的handler之后调用，无论是否存在exception都会被触发。使用示例：

    def log_request_tear_down(sender, **extra):
    
    sender.logger.debug('Reqeust is tearing down now....')

from flask import request_tearing_down
    
request_tearing_down.connect(log_request_tear_down, app)

  • flask.appcontexttearingdown

    这个信号在appcontext销毁时触发，信号的subscriber在teardown的handler之后调用，无论是否存在exception都会被触发。使用示例：

     def log_appcontext_tear_down(sender, **extra):
    
    sender.logger.debug('Appcontext is tearing down now....')

from flask import appcontext_tearing_down
    
appcontext_tearing_down.connect(log_request_tear_down, app)

  • flask.appcontext_pushed

    这个信号在appcontext加载时触发(TODO: PUSH=?加载)。这个信号可以用于临时的修改或者添加一些信息到应用程序的上下文环境。使用示例：

    from contextlib import contextmanager
    
from flask import appconext_pushed

@contextmanager
    
def user_set(app, user):
    
    def subscriber(sender, **extra):
    
        g.user = user
    
    with appcontext_pushed.connected_to(subscriber, app)
    
        yeild

with user_set(app, 'John'):
    
    with app.test_client() as c:
    
        resp = c.get('/user/me')
    
        assert resp.data == 'username=John'`

  • flask.message_flashed

    这个信号在应用程序提示一个信息时触发，信号发送的内容包含message:信息内容，category:类别

    def log_message(sender, message, catogery, **extra):
    
    sender.logger.debug('Message %s in catogery %s is flashed..', message, category)

from flask import message_flashed
    
message_flashed.connect(log_message, app)