第12章 自动化测试

测试代码应该和程序开发同步进行，通常的工作流程是：编写一部分代码，立刻编写配套的测试，运行测试确保一切正常，继续编写新功能，编写配套测试……按照这个流程不断迭代直至程序完成。

自动化测试主要分为下面三种：

• 单元测试（Unit Test）：对单独的代码块，比如函数进行测试。单元测试是自动化测试的主要形式，也是最基本的测试方式。

• 集成测试（Integration Test）：集成测试对代码单位之间的协同工作进行测试，比如测试Flask和各个 Flask扩展的集成代码。这部分的测试不容易编写，各个扩展通常会包含集成测试。在部署到云平台时，集成测试可以确保程序和云平台的各个接口正常协作。

• 用户界面测试（User Interface Test）：也被称为端对端测试或全链路测试，因为需要启动服务器并调用浏览器来完成测试，所以耗时比较长，适合用来测试复杂的页面交互，比如包含 JavaScript代码和AJAX请求等实现的功能。

简单使用

import os
import sys
import pytest

cur_path = os.path.abspath(__file__)
parent = os.path.dirname
sys.path.append(parent(parent(cur_path)))

from app import create_app

@pytest.fixture
def app():
    app = create_app('test')
    context = app.app_context()
    context.push()
    yield app
    context.pop()

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.mark.api
def test_domain_list(client):
    ''' 测试获取域名列表 '''
    params = {'cloud': 'qcloud'}
    resp = client.get('/domain/list', data=params)
    data = json.loads(resp.data)
    assert resp.status_code == 200
    assert data['state'] == 1
    assert data.get('data').get('domains')[0]['ServerName'] == 'qcloud'

登录验证

@pytest.fixture
def app():
    app = create_app('test')
    app.config['TESTING'] = True
    context = app.app_context()
    context.push()
    init_db()
    yield app
    app.config['TESTING'] = False
    db.session.remove()
    context.pop()

@pytest.fixture
def client(app):
    return app.test_client()

class AuthActions:
    def __init__(self, client):
        self._client = client
    def login(self, emp_id=1, password="123", remember_me=False):
        self._client.post(
            "/login",
            data={"emp_id": emp_id, "password": password, 'remember_me': remember_me},
            follow_redirects=True
        )
        return self._client
    def logout(self):
        return self._client.get("/logout", follow_redirects=True)

@pytest.fixture
def auth(client):
    return AuthActions(client)

@pytest.fixture
def auth_client(auth):
    return auth.login()

def test_add_page(auth_client):
    rep = auth_client.get(url_for('dpm.dpm_edit', dpm_id=1))
    assert rep.status_code == 200

app.context() 和 test request_ context()

with app.app_context():
    db.create_all()

• 使用test_request_context()方法只能构建一个全局的请求上下文环境 , 对应的URL默认为跟地址 , 你可以将自定义的路径作为第一个参数(path)传入
• 如果你想使用特定请求的request , session等请求上下文全局变量 . 可以使用with语句来调用test_cilent() , 这回在with语句前创建一个测试用的请求上下文 , 对应当前请

with app.test_cilent() as client:
    print(client.get('/hello'))
    print(request.endpoint)
    print(request.url)

测试 flask 命令

• 对于flask命令，Flask提供了 app.test_cli_runner()方法用于在测试中调用命令函数、捕捉输出。
• 对程序实例app调用 test_cli_runner()，它会返回一个 Flask cli_runner对象，使用它提供的 invoked方法调用命令，
  • 第一个参数为命令函数对象。返回 Result 对象，其中的 output 属性包含命令的输出内容。
  • 第二个参数为 命令的参数列表

import unittest
from sayhello import app, db
from sayhello.commands import forge, initdb

class SayHelloTestCase(unittest.TestCase):
    def setUp(self):
        db.create_all()
        self.client = app.test_client()
        self.runner = app.test_cli_runner()
    def tearDown(self):
        db.session.remove()
        db.drop_all()
    # 测试 forge命令
    def test_forge_command(self):
        # 触发对应命令的函数
        result = self.runner.invoke(forge)  
        self.assertIn('Created 20 fake messages.', result.output)
        self.assertEqual(Message.query.count(), 20)
    # 测试添加--count选项的 forge命令
    def test_forge_command_with_count(self):
        result = self.runner.invoke(forge, ['--count', '50'])
        self.assertIn('Created 50 fake messages.', result.output)
        self.assertEqual(Message.query.count(), 50)

if __name__ == '__main__':
    unittest.main()

编写 Flask 测试命令

import unitest
import click
from myapp import app

@app.cli.command()
def test():
    test_suite = unittest.TestLoader().discover('test')
    unittest.TextTestRunner(verbosity=2).run(test_suite)

使用 Selenium进行用户界面测试

• 对于包含较多 JavaScript代码的程序，仅仅编写单元测试是不够的，我们需要能实际测试页面加载 JavaScript后的实际交互功能。
• 需要使用一种新的测试形式 : 用户界面（User Interface , UI）测试。

pip install selenium --dev

import os
import time
import unittest

from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.keys import Keys

class UserInterfaceTestCase(unittest.TestCase):
    def setUp(self):
        os.environ['MOZ_HEADLESS'] = '1'
        self.client = webdriver.Firefox()
        time.sleep(1)
        if not self.client:
            self.skipTest('Web browser not available.')

    def tearDown(self):
        if self.client:
            self.client.quit()

    def login(self):
        self.client.get('http://localhost:5000')
        time.sleep(2)
        # navigate to login page
        self.client.find_element_by_link_text('Get Started').click()
        time.sleep(1)
        self.client.find_element_by_name('username').send_keys('grey')
        self.client.find_element_by_name('password').send_keys('123')
        self.client.find_element_by_id('login-btn').click()
        time.sleep(1)

    def test_index(self):
        self.client.get('http://localhost:5000')  # navigate to home page
        time.sleep(2)
        self.assertIn('We are todoist, we use todoism.', self.client.page_source)

    def test_login(self):
        self.login()
        self.assertIn('What needs to be done?', self.client.page_source)

    def test_new_item(self):
        self.login()
        all_item_count = self.client.find_element_by_id('all-count')
        before_count = int(all_item_count.text)
        item_input = self.client.find_element_by_id('item-input')
        item_input.send_keys('Hello, World')
        item_input.send_keys(Keys.RETURN)
        time.sleep(1)
        after_count = int(all_item_count.text)
        self.assertIn('Hello, World', self.client.page_source)
        self.assertEqual(after_count, before_count + 1)

    def test_delete_item(self):
        self.login()
        all_item_count = self.client.find_element_by_id('all-count')
        before_count = int(all_item_count.text)
        item1 = self.client.find_element_by_xpath("//span[text()='test item 1']")
        hover_item1 = ActionChains(self.client).move_to_element(item1)
        hover_item1.perform()
        delete_button = self.client.find_element_by_class_name('delete-btn')
        delete_button.click()
        after_count = int(all_item_count.text)
        self.assertNotIn('test item 1', self.client.page_source)
        self.assertIn('test item 2', self.client.page_source)
        self.assertEqual(after_count, before_count - 1)

    def test_edit_item(self):
        self.login()
        time.sleep(1)
        try:
            item = self.client.find_element_by_xpath("//span[text()='test item 1']")
            item_body = 'test item 1'
        except NoSuchElementException:
            item = self.client.find_element_by_xpath("//span[text()='test item 2']")
            item_body = 'test item 2'
        hover_item = ActionChains(self.client).move_to_element(item)
        hover_item.perform()
        edit_button = self.client.find_element_by_class_name('edit-btn')
        edit_button.click()
        edit_item_input = self.client.find_element_by_id('edit-item-input')
        edit_item_input.send_keys(' edited')
        edit_item_input.send_keys(Keys.RETURN)
        time.sleep(1)
        self.assertIn('%s edited' % item_body, self.client.page_source)

    def test_get_test_account(self):
        self.client.get('http://localhost:5000')
        time.sleep(2)
        self.client.find_element_by_link_text('Get Started').click()
        time.sleep(1)
        self.client.find_element_by_id('register-btn').click()
        self.client.find_element_by_id('login-btn').click()
        time.sleep(1)
        self.assertIn('What needs to be done?', self.client.page_source)

    def test_change_language(self):
        self.skipTest(reason='skip for materialize toast div overlay issue')
        self.login()
        self.assertIn('What needs to be done?', self.client.page_source)

        self.client.find_element_by_id('locale-dropdown-btn').click()
        # ElementClickInterceptedException: Message: Element <a class="lang-btn"> is not clickable at point
        # (1070.4000244140625,91.75) because another element <div id="toast-container"> obscures it
        self.client.find_element_by_link_text(u'简体中文').click()
        time.sleep(1)
        self.assertNotIn('What needs to be done?', self.client.page_source)
        self.assertNotIn(u'你要做些什么？', self.client.page_source)

    def test_toggle_item(self):
        self.skipTest(reason='wait for fix')
        self.login()
        all_item_count = self.client.find_element_by_id('all-count')
        active_item_count = self.client.find_element_by_id('active-count')
        before_all_count = int(all_item_count.text)
        before_active_count = int(active_item_count.text)
        self.client.find_element_by_xpath("//a[@class='done-btn'][1]").click()
        time.sleep(1)
        after_all_count = int(all_item_count.text)
        after_active_count = int(active_item_count.text)
        self.assertEqual(after_all_count, before_all_count - 1)
        self.assertEqual(after_active_count, before_active_count + 1)

    def test_clear_item(self):
        self.login()
        all_item_count = self.client.find_element_by_id('all-count')
        before_all_count = int(all_item_count.text)
        self.client.find_element_by_id('clear-btn').click()
        after_all_count = int(all_item_count.text)
        self.assertEqual(after_all_count, before_all_count - 1)

使用 Coverage.py 计算测试覆盖率

pip install coverage -- dev

coverage run maths . py

coverage run --source oa  --branch -m pytest
coverage report
coverage html -d htmlcov

• 传入 –source 选项指定要检查的包或模块为 oa
• 可选的-branch选项用来开启分支覆盖检查，比如，这会将 if 判断中未执行到的elif或else子句也视为未覆盖。

第13章 性能优化

SQL慢查询

• 使用Fask-Debug Toolbar来调试程序，其实它还内置了一个 Profiler（性能分析器）。这个 Profiler默认是关闭的，单击工具栏中 Profiler选项右上方的对号按钮来激活它。
• 记录数据库慢查询的最简单的方式是使用Fask-SQLALchemy提供的 get_debug _queries()函数。当我们开启查询记录后，在每个请求结束时调用 get_debug_queries 函数即可获得该请求所有数据库查询的信息，包括SQL语句、参数、时长等。我们可以通过将配置变量 SQLALCHEMY_RECORD_QUERIES设为True来显式地开启查询记录功能

# 一般来说，一次合理的查询不应该超过1秒，所以我们把这个值设为1秒。
BLUELOG_SLOW_QUERY_THRESHOLD = 1

def register_request_handlers(app):
    @app.after_request
    def query_profiler(response):
        for q in get_debug_queries():
            if q.duration >= app.config['BLUELOG_SLOW_QUERY_THRESHOLD']:
                app.logger.warning(
                    'Slow query: Duration: %fs\n Context: %s\nQuery: %s\n '
                    % (q.duration, q.context, q.statement)
                )
        return response

• statement : SQL语句
• parameters : 查询参数
• end_time : 结束时间
• start_time : 开始时间
• duration : 持续时间
• context : 查询所在的位置

使用 Flask-Caching 设置缓存

pip install flask-caching
pip install redis

CACHE_TYPE的值

• null（默认值） : 不使用缓存
• redis : 使用Redis
• simple : 使用本地的 Python字典
• uwsgi : 使用 uWSGI 内置的缓存框架
• filesystem : 使用文件系统存储
• memcached : 使用 Memcached

存缓视图函数

• 为视图函数附加一个 cache cached装饰器即可开启缓存
• 被缓存的数据会以键值对的形式存储起来，当下次处理请求时会先查找是否存在对应键的数据，所以我们要确保被缓存的不同值的键是唯一的。当缓存视图函数返回值时，它使用当前请求的 request path值来构建缓存数据的键，即view/%(reques.path)s。也就是说，如果URL中包含查询字符串的话，这部分内容会被忽略掉。
  • 如果想将query_string一起存缓下来的话 , 可以使用query_string=True参数

CACHE_TYPE = 'redis'

@app.route('/bar')
@cache.cached(timeout=10*60,query_string=True)
def bar():
    args = request.args.get('page')
    return render_template('bar.html')

缓存其他函数

• 和缓存视图函数相同，我们也使用 cache_cached装饰器设置缓存。
• 不同的是，你必须使用 key_prefix 关键字为缓存数据设置一个缓存键。
• 如果没有设置，Flask-Caching会使用当前请求的 request path的值，这有可能会覆盖视图函数的数据。

@cache.cached(key_perfix='add'):
def add(a,b):
    return a + b

注意 , 说是说存缓函数 , 其实是存缓函数的返回值

>>> from app import add
>>> add(1,1)    # 第一次调用 , 返回返回值被存缓
2
>>>add(2,5)     # 因为被存缓 , 所以返回值仍然是2
2
>>>add(5,5)     # 返回值依旧相同 , 知道存缓过期
2

• 上面的方法不管参数如何 , 都只是返回同一个值 , 下面的方法就会考虑传入的参数

• 使用 memorized装饰器。它的用法和 cached完全相同，

@cache.memorize()
def add_pro(a,b):
    return a + b

>>> from app import add
>>> add(1,1)    # 第一次调用 , 返回返回值被存缓
2
>>>add(2,5)     # 参数不同，再次调用函数，返回值被缓存
7
>>>add(2,5)     # 直接使用缓存，耗时小到忽略不计
7

删除存缓

• 删除视图存缓 : 调用 cache.delete()方法来清除缓存，传入特定的键来获取对应的缓存。视图函数缓存的键默认为view/<请求路径 request.path>，这里使用 url_for())函数构建缓存的键
• 删除其他函数存缓 : delete_memorized()方法来删除缓存
• cache.clear()来清除程序中的所有缓存

@app.route('/update/bar')
def update_bar():
    cache.delete_memoized(add_pro)  # 删除 add_pro的存缓
    cache.delete('view/%s' %url_for('bar')) # 删除视图函数的存缓
    cache.clear()  # 删除全部存缓
    return redirect(url_for('index'))

使用 Flask-Assets 优化静态资源

pip install flask-assets

功能 : 将一些文件需要被压缩并打包成单个文件

第14章 部署上线

Git上传空目录

• git无法追踪一个空的文件夹，当用户需要追踪(track)一个空的文件夹的时候，按照惯例，大家会把一个称为.gitkeep的文件放在这些文件夹里。
• 所以 , 我们可以在目录下创建一个.gitkeep文件 , 写入一个*.log规则 , 这样就会将logs目录添加到git , 但是忽略所有以.log结尾的日志文件

手动导入环境变量

• 在开发时，因为安装了 python-dotenv，使用 flask run命令启动开发服务器时Flask会自动导入存储在.flaskenv或.env文件中的环境变量。
• 在生产环境下，我们需要使用性能更高的生产服务器，所以不能再使用这个命令启动程序，这时我们需要手动导入环境变量。

from dotenv import load_dotenv

dotenv_path = ps.path.join(ps.path.dirname(__file__),'.env')
if os.path.exists(dotenv_path):
    load_dotenv(dotenv_path)

from bluelog import create_app
app = create_app('production')

HTTPS 转发

pip install flask-sslify

使用超级简单

from flask_sslify import SSLify

sslify = SSLify()
sslify.init_app(app)

• 强制所有发到程序的请求通过 Https，具体的方法是拦截不安全的请求并重定向到 Https。
• Flask-SSLify会自动为我们的程序处理请求。我们可以通过配置键 SSL_DISABLED来设置关闭SSL转发功能。

使用 Gunicorn 运行程序

pip install gunicorn

• 在开发时，我们使用 flask run命令启动的开发服务器是由 Werkzeug 提供的。细分的话，Werkzeug提供的这个开发服务器应该被称为 WSGI 服务器，而不是单纯意义上的Web服务器。

• Gunicorn 使用下面的命令模式来运行一个WGSI程序

  gunicorn [options] 模块名:变量名

  eg :

  gunicorn --work=4 wsgi:app

  • 这里的变量名即要运行的WSGI可调用对象，也就是我们使用Flask创建的程序实例，
  • 而模块名即包含程序实例的模块。

• 使用–workers选项来定义 worker（即工作线程）的数量。通常来说，worker的数量建议为（2×CPU核心数）+1。

使用 Supervisor管理进程

yum install -y supervisor

• 我们直接通过命令来运行 Unicorn，这并不十分可靠。
• 我们需要一个工具来自动在后台运行它，同时监控它的运行状况，并在系统出错或是重启时自动重启程序

Supervisor是一个使用 Python编写的UNX-like系统进程管理工具，它可以管理某个项目相关的所有服务。

• 安装 Supervisor后，它会自动在/etc目录下生成一个包含全局配置的配置文件名为 supervisor.conf的配置文件（IN风格语法）来定义进程相关的命令等信息。

• 和 Nginx类似，我们也可以将程序相关的配置写在这里，但是为了便于管理，我们可以为程序配置创建单独的配置文件。这个全局配置默认会将/etc/supervisor/conf.d目录下的配置文件也包含在全局配置文件中，所以我们创建一个 bluelog.conf存储程序配置：这个文件可以放在etc/supervisor.conf路径下。我们使用nano来创建这个文件

  nano /etc/supervisor/conf.d/bluelog.conf

• 配置如下:

  [program:bluelog]
  command=pipenv run gunicorn -v 4 wsgi:app
  directory=/home/greyli/bluelog
  user=heyingliang
  autostart=true
  autorestart=true
  stopasgroup=true
  killasgroup=true
  
  [supervisord]
  environment=LC_ALL='en_US.UTF-8',LANG='en_US.UTF-8'

  1. 通过 [program]定义一个 bluelog 程序，
  2. 其中用 command 定义命令，我们在命令前添加pipenv run以便在虚拟环境中执行命令；
  3. directory和user则分别用来设置执行命令的工作目录和用户；
  4. 通过将 autostart和 autostart设为true开启自动启动和自动重启；
  5. 将 stopasgroup和killasgroup设为true则会确保在关闭程序时停止所有相关的子进程

• 启动supervisor服务

  sudo service supervisor restart

第15章 Flask 扩展开发(编写flask-share)

如我们将要编写的Flask-Share扩展集成了JavaScript库share.js（https://github.com/overtrue/share.js），它的主要作用就是允许你在模板中创建社交分享（social share）组件。

Flask扩展通常分为两类：

• 一类是纯功能的实现，比如提供用户认证功能的 Flask-Login；
• 另一类是对已有的库和工具的包装，比如 Flask-SQLAlchemy就包装了 SQLAlchemy。

扩展项目骨架

一个扩展，在项目文件层面就是一个 Python开源项目。对于一个最小的项目来说，唯一必需的只有程序脚本和setup.py。

但是为了便于开发和协作，其他文件也是必不可少的。一般来说，扩展项目由下面这些文件组成

• 存储扩展代码的程序包或模块（必需）
• setup.py（必需）
• 示例程序
• 文档
• 测试脚本或包
• README（说明文档）
• LISCENCE（许可证文件）
• CHANGES（版本变更记录）
• .gitignore

编写扩展类

• 在创建的Flask扩展文件夹的根目录 , 必须要有__init__.py 这会让 flask_share变成包

• 在大多数情况下，扩展需要创建一个类来实现集成机制，并通过实例化这个类获得的扩展对象来提供主要的功能接口。在编写程序时，当我们要使用某个扩展，我们通常会实例化扩展类，并传人程序实例app以进行初始化。

• init_app()函数的基本内容 : 获取程序的配置，设置Jinja2环境，向模板上下文中添加变量或是注册各类处理函数。

  操作步骤如下:

  1. 将扩展添加到 app.extensions 属性中。

  2. 把扩展类添加到模板上下文中 : 如果需要提供用于生成HTML代码的方法，为了让这个方法可以在模板中调用

  3. 添加拓展相关配置

  4. 设置加载静态资源

  5. 设置蓝本 (注意修改static_url_path)

     因为用户通过实例化 Flask类时传入static_url_path参数可以自定义静态文件路径，这里为了和用户的设置保持一致，使用 app.static_url_path属性拼接

• 示例 :

  class OALogin(object):
      def __init__(self, app=None):
          if app is not None:
              self.init_app(app)
      def init_app(self, app):
          # 1.将扩展添加到 app.extensions 属性中
          # 这样就能使用current_app.extensions['oalogin']获取这个类 , 
          # 不过一般是通过from XX.extensions import oalogin来获取的
          if not hasattr(app,'extensions'):
              app.extensions = {}
          app.extensions['oalogin'] = self
          
          # 2.如果需要提供用于生成HTML代码的方法 , 把扩展类添加到模板上下文中
          # 这样就能通过{{ oalogin }}获取这个类 , 之后就可以调用相关函数
          app.jinja_env.globals['oalogin'] = self
          
          # 3.添加拓展相关配置
          app.config.setdefault("ZK_LOGIN_SECRET_KEY", None)
          
          # 4. 设置加载静态资源
          # 这样模板就可以通过{{ oalogin.load_css() }}获取静态文件
          @staticmethod
          def load_css(css_url=None,js_url=None):
              if css_url is None:
                  css_url = 'http://cdn.bootcss.com/socail-share.js/1.0.16/css/share.min.css'
              if js_url is None:
                  js_url = 'http://cdn.bootcss.com/social-share.js/1.016/js/socail-sharea.min.js'
              return Markup('''
              	<link rel="stylesheet" href="%s" type="text/css">\n
              	<script src="%s"></script>
              ''' %(css_url,js_url))
          
          # 5.设置蓝本
          # 注意修改static_url_path
          bp = Blueprint('share'.__name__,static_folder='static',static_url_path='/share'+app.static_url_path)
          app.register_blueprint(bp)
          
          self.configure_loginmanager(app)
          cache["redis"] = redis.StrictRedis.from_url(app.config["ZK_REDIS_CACHE_URL"])
          # 配置上下文
          @app.context_processor
          def project_name():
              return dict(site_name=app.config['SITE_NAME'])

创建 setup.py

• 为了便于分发程序，我们必须对项目进行打包（packaging），这是让你的程序可以使用pipPipenv或其他工具从PyPI安装的必要步骤。
• Python包通常使用 setuptools进行打包

setup.py文件 : 这个文件定义了 Python包的元数据，比如包的版本、名称、作者信息等。更重要的是，通过
setup.py可以对打包安装等行为进行非常详细的配置。

from setuptools import setup

setup(
        name='demo',     # 包名字
        version='1.0',   # 包版本
        description='This is a test of the setup',   # 简单描述
        author='huoty',  # 作者
        author_email='sudohuoty@163.com',  # 作者邮箱
        url='https://www.konghy.com',      # 包的主页
        packages=['demo'],                 # 包
)

更多参数可见：https://setuptools.readthedocs.io/en/latest/setuptools.html

第16章 Flask工作原理与机制解析

Flask 程序包各模块分析表

模块 / 包	说明
json/	提供 JSON 支持
__init__.py	构造文件，导入了所有其他模块中开放的类和函数
__main__.py	用来启动flask命令
_compat.py	定义 Python2 与 Python3 版本兼容代码
app.py	主脚本，实现了 WSGI 程序对象，包含 Flask 类
blueprint.py	蓝本支持，包含 Blueprint 类定义
cli.py	提供命令行支持，包含内置的几个命令
config.py	实现配置相关的对象
ctx.py	实现上下文对象，比如请求上下文 RequestContext
debughelpers.py	一些辅助开发的函数 / 类
globals.py	定义全局对象，比如 request、session 等
helpers.py	包含一些常用的辅助函数，比如 flash()、url_for()
logging.py	提供日志支持
sessions.py	实现 session 功能
signals.py	实现信号支持，定义了内置的信号
templating.py	模板渲染功能
testing.py	提供用于测试的辅助函数
views.py	提供了类似 Django 中的类视图，用于编写 Web API 的 MethodView
wrappers.py	实现 WSGI 封装对象，比如代表请求和响应的 Request 对象和 Response 对象

我们需要关注的是实现 Flask 核心功能的模块，比如WSGI交互、蓝本、上下文等。

API 文档 : https://flask.palletsprojects.com/en/1.1.x/api/

flask-origin : https://github.com/greyli/flask-origin/blob/master/flask.py

如何阅读源码

在阅读源码时，我们需要带着两个问题去读

• 这段代码实现了什么功能？
• 它是如何实现的？

• Request的最本质作用 : 用来记住匹配的端点值（endpoint）和视图参数（view arguments）
• _RequestContext : 请求上下文（request context）包含所有请求相关的信息。它会在请求进入时被创建，然后被推送到_request_ctx_stack，在请求结束时会被相应的移除。它会为提供的WSGI环境创建URL适配器（adapter）和请求对象。
• Flask : 作为一个中心注册处，所有的视图函数、URL规则、模板配置等等都将注册到这里。

显式程序对象

在一些 Python Web框架中，一个视图函数可能类似这样：

from example framework import route
@route ( ' / ' )
def index ():
return 'Hello World ! '

而在 Flask中，则需要这样 :

from flask import Flask
app = Flask(__name__)

@app.route('/')
def index():
    return 'hello world'

你应该看到其中的区别了，Flask中存在一个显式的程序对象，我们需要在全局空间中创建它。

这样设计主要有下面几个原因

• 前一种方式（隐式程序对象）在同一时间内只能有一个实例存在，而显式的程序对象允许多个程序实例存在。
• 允许你通过子类化Flask类来改变程序行为。
• Flask需要通过传入的包名称来定位资源（模板和静态文件）。
• 允许通过工厂函数来创建程序实例，可以在不同的地方传入不同的配置来创建不同的程序实例。
• 允许通过蓝本来模块化程序。

本地上下文

• 在多线程环境下，要想让所有视图函数都获取请求对象。最直接的方法就是在调用视图函数时将所有需要的数据作为参数传递进去，但这样一来程序逻辑就变得冗余且不易于维护。另一种方法是将这些数据设为全局变量，但是如果直接将请求对象设为全局变量，那么必然会在不同的线程中导致混乱（非线程安全）。
• 本地线程（thread locals）的出现解决了这些问题。
• 本地线程就是一个全局对象，你可以使用一种特定线程且线程安全的方式来存储和获取数据。也就是说，同一个变量在不同的线程内拥有各自的值，互不干扰。实现原理其实很简单，就是根据线程的ID来存取数据。Flask没有使用标准库的 threading.local()，而是使用了 Werkzeug自己实现的本地线程对象 werkzeug.local.Local()，后者增加了对 Greenlet的优先支持。

• Flask使用本地线程来让上下文代理对象全局可访问，比如 request、session、current_app、g，这些对象被称为本地上下文对象（context locals）。因此，在不基于线程、greenlet或单进程实现的并发服务器上，这些代理对象将无法正常工作。
• Flask的设计初衷是为了让传统Web程序的开发更加简单和迅速，而不是用来开发大型程序或异步服务器的。但是 Flask的可扩展性却提供了无限的可能性，除了使用扩展，我们还可以子类化 Flask类，或是为程序添加中间件。

三种程序状态

Flask提供的四个本地上下文对象分别在特定的程序状态下绑定实际的对象。如果我们在访问或使用它们时还没有绑定，那么就会看到初学者经常见到的 Runtime error异常。

在 Flask中存在三种状态，分别是

1. 程序设置状态（application setup state）、
2. 程序运行状态（application runtime state）
3. 请求运行状态（request runtime state）。

程序设置状态

当 Flask类被实例化，也就是创建程序实例app后，就进入了程序设置状态。

这时所有的全局对象都没有被绑定

from flask import Flask , current_app , g , request , session
app = Flask(__name__)

print(current_app , g , request , session)
# (
# <LocalProxy unbound>,
# <LocalProxy unbound>,
# <LocalProxy unbound>,
# <LocalProxy unbound>,
# )

程序运行状态

• 当 Flask程序启动，但是还没有请求进入时，Flask进入了程序运行状态。在这种状态下，程序上下文对象 current_app和 g 都绑定了各自的对象。
• 使用 flask shell 命令打开的 Python shell 默认就是这种状态，
• 我们也可以在普通的 Python shell中通过手动推送程序上下文来模拟

from flask import Flask , current_app , g , request , session
app = Flask(__name__)
ctx = app.app_context()
ctx.push()

print(current_app , g , request , session)
# (
# <Flask '__main__'>,
# <Flask.g '__main__'>,
# <LocalProxy unbound>,
# <LocalProxy unbound>,
# )
ctx.pop()

• 在上面的代码中，我们手动使用 app_context()方法创建了程序上下文，然后调用push())方法把它推送到程序上下文堆栈里。
• 当请求进入的时候，程序上下文会随着请求上下文一起被自动激活。
• 但是在没有请求进入的场景，比如离线脚本、测试，或是进行交互式调试的时候，手动推送程序上下文以进入程序运行状态会非常方便。

请求运行状态

• 当请求进入的时候，或是使用 test_request_context()方法、test_client()方法时，Flask会进人请求运行状态。
• 因为当请求上下文被推送时，程序上下文也会被自动推送，所以在这个状态下4个全局对象都会被绑定，
• 我们可以通过手动推送请求上下文模拟：

from flask import Flask , current_app , g , request , session
app = Flask(__name__)
ctx = app.test_request_context() 
ctx.push()

print(current_app , g , request , session)
# (
# <Flask '__main__'>,
# <Flask.g '__main__'>,
# <Request 'http://localhost/' [GET]>,
# <NullSession {}>,
# )
ctx.pop()

这也是为什么你可以直接在视图函数和相应的回调函数里直接使用这些上下文对象，而不用推送上下文——Flask在处理请求时会自动帮你推送请求上下文和程序上下文。

自定义

子类化Flask对象

from flask import Flask

class MyFlask(Flask):
    pass

app = MyFlask(__name__)

...

子类化Request对象

from flask import Flask, Request

class MyRequest(Request):
    pass

app = Flask(__name__)
app.request_class = MyRequest

...

子类化Response对象

from flask import Flask, Response

class MyResponse(Response):
    pass

app = Flask(__name__)
app.response_class = MyResponse

...

Flask与WSGI

• Python Web Server Gateway Interface，它是为了让web服务器与 Python 程序能够进行数据交流而定义的一套接口标准/规范。
• 如果不统一标准，那么众多的 Python Web框架都可能仅被某些Web服务器支持；而Web服务器也没法支持所有的 Python Web框架。
• Flask的核心扩展 Werkzeug是一个WSGI工具库。

WSGI 程序

根据WSGI的规定，Web程序（或被称为WSGI程序）必须是一个可调用对象。这个可调用对象接收两个参数

• environ：包含了请求的所有信息的字典。
• start_response：需要在可调用对象中调用的函数，用来发起响应，参数是状态码、响应头部等。

WSGI服务器会在调用这个可调用对象时传入这两个参数。另外，这个可调用对象还要返回个可迭代（iterable）的对象。

# 一个最简单的WSGI程序 : func形式
def hello(environ, start_response):
    status = '200 OK'
    response_headers = [('Content-type', 'text/html')]
    start_response(status , response_headers)
    return [b'<h1> Hello , Web</h1>']

# 或者
# 一个最简单的WSGI程序 : class形式
class AppClass:
    def __init__(self,environ,start_response):
        self.environ = environ
        self.start = start_resopnse
        
    def __iter__(self):
        status = '200 OK'
        response_headers = [('Content-type', 'text/html')]
        self.start(status , response_headers)
        yield b'<h1> Hello , Web</h1>'
        
# 如果想以类的实例作为WSGI程序，那么这个类必须实现__call__方法。

• 如果想以类的实例作为WSGI程序，那么这个类必须实现__call__方法。
• 在 Flask中，这个可调用对象就是我们的程序实例app，我们创建app实例时调用的 Flask 类就是另一种可调用对象形式——实现了__call__方法的类

WSGI 服务器

程序编写好了，现在我们需要一个WSGI服务器来运行它。

from wsgiref.simple_server import make_server

def hello(environ, start_response):
    status = '200 OK'
    response_headers = [('Content-type', 'text/html')]
    start_response(status , response_headers)
    return [b'<h1> Hello , Web</h1>']

# 创建一个本地服务器，分别传人主机地址、端口和可调用对象(WSGI程序)
server = make_server('localhost',5000,hello)
# 运行WSGI服务器
server.server_forever()

• WSGI服务器启动后，它会监听本地机的对应端口（我们设置的5000）。
• 当接收到请求时它会把请求报文解析为一个 environ 字典，然后调用WSGI程序提供的可调用对象，传递这个字典作为参数，同时传递的另一个参数是一个 start response函数。
• 所以 , Flask提供的请求对象Request其实就是对 environ字典的解析和封装。

Flask 的工作流程与机制

1. 程序启动
   1. 调用fask. cli_run_command()函数
   2. fask.cli_run_command()函数 调用run_simple() 函数
   3. run_simple() 根据DEBUG模式 , 决定是否调用调试器和重载器 , 然后调用serve_forever() , 运行服务器
2. 请求 In
   1. 当请求进来时 , 调用Flask类的__call__方法，
   2. __call__方法内部调用了Flask.wsgi_app()方法
   3. Flask.wsgi_app()首先尝试从 Flask.full_dispatch_request()方法获取响应，如果出错那么就根据错误类型来生成错误响应。
   4. full_dispatch_request()方法首先执行preprocess_request()方法对请求进行预处理(指向所有实用brefore_request钩子注册的函数) , 接着调用dispatch_request()方法匹配并调用对应的视图函数，获取其返回值 , 最后使用finalize_request()方法生成响应
3. 响应 Out
   1. finalize_request()方法调用make_response()方法 , 接着执行process_response()方法会响应进行预处理
   2. 返回response结果回Flask.wsgi_app()方法

路由系统

注册路由

路由表 Map

m = Map()
rule1 = Rule('/',endpoint='index')
m.add(rule1)

route函数

def route(self, rule, **options):
    """route函数简化版 """
    def decorator(f):
        self.add_url_rule(rule, f.__name__, **options)  # add_url_rule()背后调用url_map.add(rule)
        # 将端点（默认使用函数名，即f.__name__）和函数对象的映射存储到view_functions字典
        self.view_functions[f.__name__] = f  
        return f
    return decorator

1. url_map 是 Werkzeug的Map类实例（werkzeug.routing.Map）。它存储了URL规则和相关配置，这里的rule是 Werkzeug提供的Rule实例（werkzeug.routing.Rule），其中保存了端点和URL规则的映射关系。
2. view function则是 Flask类中定义的一个字典，它存储了端点和视图函数的映射关系。

路由匹配

m = Map()
rule1 = Rule('/',endpoint='index')
m.add(rule1)

urls = m.bind('example.com')
res1 = urls.match('/','GET')
res2 = urls.match('/downloads/42','GET')

print(res1)  # ('index',{})
print(res2)  # ('/downloads/show',{'id':42})

Map.bind() 方法和 Map.bind_to_environ()都会返回一个 Map Adapter对象 , 它负责匹配和构建URL.

MapAdapter类的 match方法用来判断传入的URL是否匹配Map对象中存储的路由规则(存储在 self.map.rules列表中).匹配成功后会返回一个包含URL端点和URL变量的元组.

本地上下文

• 本地线程（Thread Local）: 在保存数据的同时记录下对应的线程ID，获取数据时根据所在线程的ID即可获取到对应的数据。就像是超市里的存包柜，每个柜子都有一个号码，每个号码对应一份物品。
• 本地线程自建了一个__storege__属性用于存放数据 , 其数据结构为{线程ID : {名称 : 实际数据}}

from functools import partial

from werkzeug.local import LocalProxy
from werkzeug.local import LocalStack

# 两个错误信息
_request_ctx_err_msg = """Working outside of request context."""
_app_ctx_err_msg = """Working outside of application context."""

# 查找请求上下文对象
def _lookup_req_object(name):
    top = _request_ctx_stack.top
    if top is None:
        raise RuntimeError(_request_ctx_err_msg)
    return getattr(top, name)

# 查找程序上下文对象
def _lookup_app_object(name):
    top = _app_ctx_stack.top
    if top is None:
        raise RuntimeError(_app_ctx_err_msg)
    return getattr(top, name)

# 查找程序实例
def _find_app():
    top = _app_ctx_stack.top
    if top is None:
        raise RuntimeError(_app_ctx_err_msg)
    return top.app

# 两个堆栈
_request_ctx_stack = LocalStack()  # 请求上下文堆栈
_app_ctx_stack = LocalStack()  # 程序上下文堆栈

# 4个全局上下文代理对象
current_app = LocalProxy(_find_app)
request = LocalProxy(partial(_lookup_req_object, "request"))
session = LocalProxy(partial(_lookup_req_object, "session"))
g = LocalProxy(partial(_lookup_app_object, "g"))

• 简单来说，Localstack是基于 Local实现的栈结构（本地堆栈，即实现了本地线程的堆栈），有 push、pop0方法以及获取栈顶的top属性。
• 在构造函数中创建了 Local()类的实例 local。它把数据存储到 Local中，并将数据的字典名称设为'stack'。
• 注意这里和 Local类一样也定义了call方法，当 Localstack实例被直接调用时，会返回栈顶对象的代理，即 LocalProxy类实例

代理与LocalProxy

• 代理（Proxy）是一种设计模式，
• 通过创建一个代理对象 , 我们可以使用这个代理对象来操作实际对象。
• 从字面理解，代理就是使用一个中间人来转发操作。

class Proxy:
    def __init__(self , obj):
        object.__setattr__(self , '_obj' , obj)
    
    def __getattr__(self , name):
        return getattr(self._obj , name)
    
    def __setattr__(self, name, value):
        self._obj[name] = value
        
    def __delattr__(self, name):
        del self._obj[name]

为什么 Flask需要使用代理？

• 总体来说，在这里使用代理对象是因为这些代理可以在线程间共享，让我们可以以动态的方式获取被代理的实际对象。
• 具体来说，Flask的三种状态中，当上下文没被推送时，响应的全局代理对象处于未绑定状态。而如果这里不使用代理，那么在导入这些全局对象时就会尝试获取上下文，然而这时堆栈是空的，所以获取到的全局对象只能是None。当请求进入并调用视图函数时，虽然这时堆栈里已经推入了上下文，但这里导入的全局对象仍然是None。总而言之，上下文的推送和移除是动态进行的，而使用代理可以让我们拥有动态获取上下文对象的能力。
• 另外，一个动态的全局对象，也让多个程序实例并存有了可能。这样在不同的程序上下文环境中，current_app总是能对应正确的程序实例。