教程(WSGI)

在本教程中,我们将介绍如何为简单的图像共享服务构建API。在此过程中,我们将讨论Falcon的主要特性,并介绍框架使用的术语。

备注

本教程介绍了Falcon的“传统”、同步风格,它使用 WSGI 协议。

开发一个 async 申请?请查看我们的 ASGI tutorial 取而代之的是!

第一步

我们要做的第一件事是 install Falcon在新鲜的 virtualenv .为此,我们创建一个名为“look”的新项目文件夹,并在其中设置一个可用于教程的虚拟环境:

$ mkdir look
$ cd look
$ virtualenv .venv
$ source .venv/bin/activate
$ pip install falcon

通常情况下,项目的顶级模块被称为与项目相同,因此,让我们在第一个模块内创建另一个“look”文件夹,并通过创建一个空的 __init__.py 其中的文件:

$ mkdir look
$ touch look/__init__.py

接下来,让我们创建一个新的文件,作为进入应用程序的入口点:

$ touch look/app.py

文件层次结构现在应该如下所示:

look
├── .venv
└── look
    ├── __init__.py
    └── app.py

现在,打开 app.py 在您最喜欢的文本编辑器中添加以下行:

import falcon

app = application = falcon.App()

此代码创建您的WSGI应用程序,并将其别名为 app 。您可以使用任何您喜欢的变量名称,但我们将使用 application 因为这是Gunicorn默认期望的名称(我们将在本教程的下一节中看到这是如何工作的)。

备注

wsgi应用程序只是一个具有定义良好的签名的可调用应用程序,这样您就可以用任何了解 WSGI protocol .

接下来,让我们看看 falcon.App 类。安装 IPython 然后启动它:

$ pip install ipython
$ ipython

现在,输入以下内容来反省 falcon.App 可调用的:

In [1]: import falcon

In [2]: falcon.App.__call__?

或者,您可以使用标准的python help() 功能:

In [3]: help(falcon.App.__call__)

注意方法签名。 envstart_response 是标准的wsgi参数。Falcon在这些参数上添加了一个很薄的抽象,因此您不必直接与它们交互。

Falcon框架包含大量的内联文档,您可以使用上述技术进行查询。

小技巧

除了 IPython ,python社区维护了一些您可能希望尝试的其他超能力repl,包括 bpythonptpython .

托管应用程序

现在您有了一个简单的falcon应用程序,您可以使用wsgi服务器进行旋转。python包含一个用于自托管的参考服务器,但是让我们使用一些在生产中可能使用的更健壮的服务器。

打开新终端并运行以下操作:

$ source .venv/bin/activate
$ pip install gunicorn
$ gunicorn --reload look.app

(注意 --reload 选项告诉Gunicorn在其代码更改时重新加载应用程序。)

如果您是Windows用户,则可以使用服务生代替Gunicorn,因为后者不在Windows下工作:

$ pip install waitress
$ waitress-serve --port=8000 look.app:app

现在,在另一个终端中,尝试使用curl查询跑步应用程序:

$ curl -v localhost:8000

你应该得到404。实际上没关系,因为我们还没有指定任何路线。Falcon包含一个默认的404响应处理程序,它将为路由不存在的任何请求路径触发。

虽然curl确实完成了任务,但使用起来可能有点粗糙。 HTTPie 是一个现代的,用户友好的选择。让我们安装httpie并从现在开始使用它:

$ source .venv/bin/activate
$ pip install httpie
$ http localhost:8000

创建资源

Falcon的设计借鉴了其他建筑风格的几个关键概念。

休息和Falcon框架的核心是“资源”的概念。资源只是API或应用程序中所有可以通过URL访问的东西。例如,一个事件预订应用程序可能有“门票”和“场地”等资源,而一个视频游戏后端可能有“成就”和“玩家”等资源。

URL为客户机提供了一种唯一标识资源的方法。例如, /players 可以识别“所有玩家列表”资源,同时 /players/45301f54 可以识别“ID为45301F54的个人玩家”,以及 /players/45301f54/achievements “ID为45301F54的播放机资源的所有成就列表”。

  POST        /players/45301f54/achievements
└──────┘    └────────────────────────────────┘
 Action            Resource Identifier

在REST体系结构样式中,URL只标识资源;它不指定对该资源采取什么操作。相反,用户从一组标准方法中进行选择。对于HTTP,这些是常见的GET、POST、HEAD等。客户机可以查询资源以发现它支持哪些方法。

备注

这是REST和RPC体系结构样式之间的主要区别之一。REST在任意数量的资源上应用一组标准的动词,而不是让每个应用程序定义自己唯一的一组方法。

根据请求的操作,服务器可能会或可能不会向客户机返回表示。表示可以编码为许多Internet媒体类型中的任何一种,例如JSON和HTML。

Falcon使用Python类来表示资源。实际上,这些类在应用程序中充当控制器。它们将一个传入的请求转换为一个或多个内部操作,然后根据这些操作的结果组合一个对客户机的响应。

           ┌────────────┐
request  → │            │
           │ Resource   │ ↻ Orchestrate the requested action
           │ Controller │ ↻ Compose the result
response ← │            │
           └────────────┘

falcon中的资源只是一个常规的python类,它包含一个或多个表示该资源支持的标准HTTP谓词的方法。每个请求的URL都映射到特定的资源。

既然我们正在构建一个图像共享API,那么让我们从创建一个“图像”资源开始。创建新模块, images.py 旁边 app.py ,并添加以下代码:

import json

import falcon


class Resource:

    def on_get(self, req, resp):
        doc = {
            'images': [
                {
                    'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
                }
            ]
        }

        # Create a JSON representation of the resource
        resp.text = json.dumps(doc, ensure_ascii=False)

        # The following line can be omitted because 200 is the default
        # status returned by the framework, but it is included here to
        # illustrate how this may be overridden as needed.
        resp.status = falcon.HTTP_200

正如你所看到的, Resource 只是一个普通班。你可以随意给班级命名。Falcon使用duck类型,因此您不需要从任何类型的特殊基类继承。

上面的图像资源定义了一个方法, on_get() .对于您希望资源支持的任何HTTP方法,只需添加一个 on_*() 类的方法,其中 * 是标准HTTP方法中的任何一种,基于较低的(例如, on_get()on_put()on_head() 等)。

备注

支持的HTTP方法是在 RFC 7231RFC 5789 .这包括get、head、post、put、delete、connect、options、trace和patch。

我们称这些著名的方法为“响应者”。每个响应程序接受(至少)两个参数,一个代表HTTP请求,另一个代表对该请求的HTTP响应。按照惯例,这些被称为 reqresp ,分别。路由模板和钩子可以注入额外的参数,稍后我们将看到。

现在,图像资源用一个简单的 200 OK 还有一具JSON尸体。Falcon的Internet媒体类型默认为 application/json 但你可以把它设置成你喜欢的任何东西。值得注意的JSON替代方案包括 YAMLMessagePack .

接下来,让我们连接这个资源并查看它的实际运行情况。返回 app.py 修改它,使它看起来像这样:

import falcon

from .images import Resource


app = application = falcon.App()

images = Resource()
app.add_route('/images', images)

现在,当一个请求 /images ,falcon将调用对应于请求的HTTP方法的图像资源上的响应程序。

让我们试试看。重新启动Gunicorn(除非使用 --reload )并向资源发送GET请求:

$ http localhost:8000/images

你应该得到一个 200 OK 响应,包括“images”资源的JSON编码表示。

备注

add_route() 需要资源类的实例,而不是类本身。相同的实例用于所有请求。此策略提高了性能并减少了内存使用,但这也意味着如果您用线程化的Web服务器托管应用程序,那么资源及其依赖关系必须是线程安全的。

我们可以使用 检查模块 要可视化应用程序配置,请执行以下操作:

falcon-inspect-app look.app:app

这将打印以下内容,正确地表明我们正在处理 GET 中的请求 /images 路线:

Falcon App (WSGI)
• Routes:
    ⇒ /images - Resource:
       └── GET - on_get

到目前为止,我们只实现了GET的响应程序。让我们看看当请求不同的方法时会发生什么:

$ http PUT localhost:8000/images

这次你应该回去 405 Method Not Allowed ,因为资源不支持 PUT 方法。注意allow头的值:

allow: GET, OPTIONS

这是由Falcon根据目标资源实现的一组方法自动生成的。如果一个资源不包括它自己的选项响应程序,框架将提供一个默认实现。因此,选项总是包含在允许的方法列表中。

备注

如果您对其他python web框架有很多经验,您可能会习惯于使用装饰器来设置路由。Falcon的特殊方法提供了以下好处:

  • 应用程序的URL结构是集中的。这使得随着时间的推移理解和维护API变得更加容易。

  • 资源类的使用自然地映射到REST体系结构样式,其中URL仅用于标识资源,而不是对该资源执行的操作。

  • 资源类方法提供了一个统一的接口,不必从一个类到另一个类、从一个应用程序到另一个应用程序进行重新设计(和维护)。

接下来,为了好玩,我们修改一下我们的资源 MessagePack 而不是JSON。从安装相关软件包开始:

$ pip install msgpack-python

然后,更新响应程序以使用新的媒体类型:

import falcon

import msgpack


class Resource:

    def on_get(self, req, resp):
        doc = {
            'images': [
                {
                    'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
                }
            ]
        }

        resp.data = msgpack.packb(doc, use_bin_type=True)
        resp.content_type = falcon.MEDIA_MSGPACK
        resp.status = falcon.HTTP_200

请注意 resp.data 代替 resp.text 。如果您将字节字符串分配给后者,Falcon会解决这个问题,但是您可以通过直接将 resp.data

同时注意使用 falcon.MEDIA_MSGPACK . 这个 falcon 模块为常见的媒体类型提供许多常量,包括 falcon.MEDIA_JSONfalcon.MEDIA_MSGPACKfalcon.MEDIA_YAMLfalcon.MEDIA_XMLfalcon.MEDIA_HTMLfalcon.MEDIA_JSfalcon.MEDIA_TEXTfalcon.MEDIA_JPEGfalcon.MEDIA_PNGfalcon.MEDIA_GIF .

重新启动Gunicorn(除非使用 --reload ,然后尝试向修订后的资源发送GET请求:

$ http localhost:8000/images

测试应用程序

充分运用代码对于创建一个健壮的应用程序至关重要。让我们花点时间来为目前已经实现的内容编写一个测试。

首先,创建 tests 目录 __init__.py 和测试模块 (test_app.py )在里面。项目的结构现在应该如下所示:

look
├── .venv
├── look
│   ├── __init__.py
│   ├── app.py
│   └── images.py
└── tests
    ├── __init__.py
    └── test_app.py

Falcon支架 testing 它的 App 通过模拟HTTP请求来实现。

Tests can either be written using Python's standard unittest module, or with any of a number of third-party testing frameworks, such as pytest. For this tutorial we'll use pytest since it allows for more pythonic test code as compared to the JUnit-inspired unittest module.

我们先安装 pytest 包裹:

$ pip install pytest

下一步,编辑 test_app.py 如下所示:

import falcon
from falcon import testing
import msgpack
import pytest

from look.app import app


@pytest.fixture
def client():
    return testing.TestClient(app)


# pytest will inject the object returned by the "client" function
# as an additional parameter.
def test_list_images(client):
    doc = {
        'images': [
            {
                'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
            }
        ]
    }

    response = client.simulate_get('/images')
    result_doc = msgpack.unpackb(response.content, raw=False)

    assert result_doc == doc
    assert response.status == falcon.HTTP_OK

在主项目目录中,通过对 tests 目录:

$ pytest tests

如果Pytest报告了任何错误,请在继续本教程的下一节之前花点时间来修复它们。

请求和响应对象

资源中的每个响应程序都会收到 Request 对象,可用于读取请求的头、查询参数和主体。你可以用这个标准 help() 函数或伊普生的魔法 ? 函数列出Falcon的属性和方法 Request 班级:

In [1]: import falcon

In [2]: falcon.Request?

每个响应者也会收到一个 Response 可用于设置响应的状态代码、头和正文的对象:

In [3]: falcon.Response?

当在应用程序中创建一个可以向集合中添加新图像资源的后端点时,这将非常有用。接下来我们将讨论这个功能。

我们这次将使用TDD来演示如何在开发Falcon应用程序时应用这个特定的测试策略。通过测试,我们将首先精确地定义应用程序要做什么,然后编写代码,直到测试告诉我们已经完成了。

备注

要了解有关TDD的更多信息,您可能希望查阅有关此主题的众多书籍之一,例如 Test Driven Development with Python .本书中的示例使用了django框架,甚至是javascript,但作者介绍了一些广泛适用的测试原则。

让我们从添加一个附加的import语句开始 test_app.py .我们需要从 unittest.mock

from unittest.mock import mock_open, call

现在添加以下测试:

# "monkeypatch" is a special built-in pytest fixture that can be
# used to install mocks.
def test_posted_image_gets_saved(client, monkeypatch):
    mock_file_open = mock_open()
    monkeypatch.setattr('io.open', mock_file_open)

    fake_uuid = '123e4567-e89b-12d3-a456-426655440000'
    monkeypatch.setattr('uuid.uuid4', lambda: fake_uuid)

    # When the service receives an image through POST...
    fake_image_bytes = b'fake-image-bytes'
    response = client.simulate_post(
        '/images',
        body=fake_image_bytes,
        headers={'content-type': 'image/png'}
    )

    # ...it must return a 201 code, save the file, and return the
    # image's resource location.
    assert response.status == falcon.HTTP_CREATED
    assert call().write(fake_image_bytes) in mock_file_open.mock_calls
    assert response.headers['location'] == '/images/{}.png'.format(fake_uuid)

正如您所看到的,这个测试很大程度上依赖于模拟,这使得它在面对实现更改时有些脆弱。我们稍后再讨论。现在,再次运行测试并观察以确保它们失败。TDD工作流程中的一个关键步骤是验证您的测试 在继续执行之前通过:

$ pytest tests

为了使新的测试通过,我们需要添加一种新的处理岗位的方法。正常开放 images.py 并将后响应者添加到 Resource 分类如下:

import io
import os
import uuid
import mimetypes

import falcon
import msgpack


class Resource:

    _CHUNK_SIZE_BYTES = 4096

    # The resource object must now be initialized with a path used during POST
    def __init__(self, storage_path):
        self._storage_path = storage_path

    # This is the method we implemented before
    def on_get(self, req, resp):
        doc = {
            'images': [
                {
                    'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
                }
            ]
        }

        resp.data = msgpack.packb(doc, use_bin_type=True)
        resp.content_type = falcon.MEDIA_MSGPACK
        resp.status = falcon.HTTP_200

    def on_post(self, req, resp):
        ext = mimetypes.guess_extension(req.content_type)
        name = '{uuid}{ext}'.format(uuid=uuid.uuid4(), ext=ext)
        image_path = os.path.join(self._storage_path, name)

        with io.open(image_path, 'wb') as image_file:
            while True:
                chunk = req.stream.read(self._CHUNK_SIZE_BYTES)
                if not chunk:
                    break

                image_file.write(chunk)

        resp.status = falcon.HTTP_201
        resp.location = '/images/' + name

如您所见,我们为图像生成一个唯一的名称,然后通过读取 req.stream .它叫 stream 而不是 body 为了强调这样一个事实,您实际上是在从输入流中读取数据;默认情况下,Falcon不假脱机或解码请求数据,而是让您直接访问wsgi服务器提供的传入二进制流。

注意使用 falcon.HTTP_201 将响应状态设置为“201已创建”。我们也可以用 falcon.HTTP_CREATED 别名。对于预定义状态字符串的完整列表,只需调用 help()falcon.status_codes

In [4]: help(falcon.status_codes)

最后一行 on_post() 响应程序为新创建的资源设置位置头。(我们将在一分钟内为该路径创建一个路由。)该 RequestResponse 类包含用于读取和设置公共头的方便属性,但您始终可以使用 req.get_header()resp.set_header() 方法。

花点时间再次运行pytest以检查进度:

$ pytest tests

你应该看到 TypeError 由于添加了 storage_path 参数到 Resource.__init__() .

要解决此问题,只需编辑 app.py 并传入初始值设定项的路径。现在,只需使用启动服务的工作目录:

images = Resource(storage_path='.')

再次尝试运行测试。这一次,他们应该以飞扬的色彩通过!

$ pytest tests

最后,重新启动gunicorn,然后尝试从命令行向资源发送post请求(替换 test.png 对于任何您喜欢的PNG路径。)

$ http POST localhost:8000/images Content-Type:image/png < test.png

现在,如果您检查存储目录,它应该包含您刚刚发布的图像的副本。

向上向前!

可测试性重构

早些时候,我们指出,我们的后期测试主要依赖于模拟,依赖于随着代码的发展可能会或可能不会成立的假设。为了缓解这个问题,我们不仅需要重构测试,还需要重构应用程序本身。

我们将从资源的后响应者分解出业务逻辑开始 images.py 以便能够独立测试。在这种情况下,资源的“业务逻辑”只是图像保存操作:

import io
import mimetypes
import os
import uuid

import falcon
import msgpack


class Resource:

    def __init__(self, image_store):
        self._image_store = image_store

    def on_get(self, req, resp):
        doc = {
            'images': [
                {
                    'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
                }
            ]
        }

        resp.data = msgpack.packb(doc, use_bin_type=True)
        resp.content_type = falcon.MEDIA_MSGPACK
        resp.status = falcon.HTTP_200

    def on_post(self, req, resp):
        name = self._image_store.save(req.stream, req.content_type)
        resp.status = falcon.HTTP_201
        resp.location = '/images/' + name


class ImageStore:

    _CHUNK_SIZE_BYTES = 4096

    # Note the use of dependency injection for standard library
    # methods. We'll use these later to avoid monkey-patching.
    def __init__(self, storage_path, uuidgen=uuid.uuid4, fopen=io.open):
        self._storage_path = storage_path
        self._uuidgen = uuidgen
        self._fopen = fopen

    def save(self, image_stream, image_content_type):
        ext = mimetypes.guess_extension(image_content_type)
        name = '{uuid}{ext}'.format(uuid=self._uuidgen(), ext=ext)
        image_path = os.path.join(self._storage_path, name)

        with self._fopen(image_path, 'wb') as image_file:
            while True:
                chunk = image_stream.read(self._CHUNK_SIZE_BYTES)
                if not chunk:
                    break

                image_file.write(chunk)

        return name

让我们检查一下,看看我们是否破坏了上面的变化:

$ pytest tests

嗯,看来我们忘了更新了 app.py .现在我们来做:

import falcon

from .images import ImageStore, Resource


app = application = falcon.App()

image_store = ImageStore('.')
images = Resource(image_store)
app.add_route('/images', images)

让我们再试一次:

$ pytest tests

现在您应该看到一个失败的测试断言 mock_file_open .为了解决这个问题,我们需要将我们的策略从猴子补丁切换到依赖注入。返回 app.py 并将其修改为如下所示:

import falcon

from .images import ImageStore, Resource


def create_app(image_store):
    image_resource = Resource(image_store)
    app = falcon.App()
    app.add_route('/images', image_resource)
    return app


def get_app():
    image_store = ImageStore('.')
    return create_app(image_store)

如您所见,大部分设置逻辑已移动到 create_app() ,它可用于获取应用程序对象,用于测试或在生产中托管。 get_app() 负责实例化其他资源并配置用于托管的应用程序。

运行应用程序的命令是:

$ gunicorn --reload 'look.app:get_app()'

最后,我们需要更新测试代码。修改 test_app.py 与此类似:

import io
from wsgiref.validate import InputWrapper

from unittest.mock import call, MagicMock, mock_open

import falcon
from falcon import testing
import msgpack
import pytest

import look.app
import look.images


@pytest.fixture
def mock_store():
    return MagicMock()


@pytest.fixture
def client(mock_store):
    app = look.app.create_app(mock_store)
    return testing.TestClient(app)


def test_list_images(client):
    doc = {
        'images': [
            {
                'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
            }
        ]
    }

    response = client.simulate_get('/images')
    result_doc = msgpack.unpackb(response.content, raw=False)

    assert result_doc == doc
    assert response.status == falcon.HTTP_OK


# With clever composition of fixtures, we can observe what happens with
# the mock injected into the image resource.
def test_post_image(client, mock_store):
    file_name = 'fake-image-name.xyz'

    # We need to know what ImageStore method will be used
    mock_store.save.return_value = file_name
    image_content_type = 'image/xyz'

    response = client.simulate_post(
        '/images',
        body=b'some-fake-bytes',
        headers={'content-type': image_content_type}
    )

    assert response.status == falcon.HTTP_CREATED
    assert response.headers['location'] == '/images/{}'.format(file_name)
    saver_call = mock_store.save.call_args

    # saver_call is a unittest.mock.call tuple. It's first element is a
    # tuple of positional arguments supplied when calling the mock.
    assert isinstance(saver_call[0][0], InputWrapper)
    assert saver_call[0][1] == image_content_type

如你所见,我们已经重做了这篇文章。虽然模拟的数量较少,但是断言已经变得更加复杂,以便正确地检查接口边界上的交互。

让我们检查一下进度:

$ pytest tests

全绿色!但是因为我们使用了模拟,所以我们不再覆盖图像的实际保存。让我们为此添加一个测试:

def test_saving_image(monkeypatch):
    # This still has some mocks, but they are more localized and do not
    # have to be monkey-patched into standard library modules (always a
    # risky business).
    mock_file_open = mock_open()

    fake_uuid = '123e4567-e89b-12d3-a456-426655440000'
    def mock_uuidgen():
        return fake_uuid

    fake_image_bytes = b'fake-image-bytes'
    fake_request_stream = io.BytesIO(fake_image_bytes)
    storage_path = 'fake-storage-path'
    store = look.images.ImageStore(
        storage_path,
        uuidgen=mock_uuidgen,
        fopen=mock_file_open
    )

    assert store.save(fake_request_stream, 'image/png') == fake_uuid + '.png'
    assert call().write(fake_image_bytes) in mock_file_open.mock_calls

现在试一试:

$ pytest tests -k test_saving_image

和以前的测试一样,这个测试仍然使用模拟。但是,通过组件化和依赖倒置技术,代码结构得到了改进,使应用程序更加灵活和可测试。

小技巧

检查代码 coverage 这将有助于我们检测上面缺失的测试;在您的工作流中包含覆盖率测试,以确保没有任何错误隐藏在未执行的代码路径中。

功能测试

功能测试从外部定义应用程序的行为。在使用TDD时,与较低级别的单元测试相比,这可能是一个更自然的开始位置,因为在定义应用程序面向用户的功能之前,很难预测需要哪些内部接口和组件。

对于上一节中的重构工作,我们可能无意中在应用程序中引入了一个我们的单元测试不会捕获的功能缺陷。当错误是多个单元之间、应用程序和Web服务器之间或应用程序与其依赖的任何外部服务之间的意外交互的结果时,就会发生这种情况。

使用测试助手,例如 simulate_get()simulate_post() ,我们可以创建跨越多个单元的测试。但我们也可以更进一步,将应用程序作为一个正常的、独立的进程(例如,使用Gunicorn)运行。然后,我们可以编写测试,这些测试通过HTTP与正在运行的进程交互,其行为类似于正常的客户机。

让我们看看这一点。创建新的测试模块, tests/test_integration.py 包括以下内容:

import os

import requests


def test_posted_image_gets_saved():
    file_save_prefix = '/tmp/'
    location_prefix = '/images/'
    fake_image_bytes = b'fake-image-bytes'

    response = requests.post(
        'http://localhost:8000/images',
        data=fake_image_bytes,
        headers={'content-type': 'image/png'}
    )

    assert response.status_code == 201
    location = response.headers['location']
    assert location.startswith(location_prefix)
    image_name = location.replace(location_prefix, '')

    file_path = file_save_prefix + image_name
    with open(file_path, 'rb') as image_file:
        assert image_file.read() == fake_image_bytes

    os.remove(file_path)

接下来,安装 requests 打包(根据新测试的要求)并确保Gunicorn已启动并运行:

$ pip install requests
$ gunicorn 'look.app:get_app()'

然后,在另一个终端中,尝试运行新测试:

$ pytest tests -k test_posted_image_gets_saved

测试将失败,因为它期望图像文件位于 /tmp .要修复此问题,请修改 app.py 要添加使用环境变量配置映像存储目录的功能,请执行以下操作:

import os

import falcon

from .images import ImageStore, Resource


def create_app(image_store):
    image_resource = Resource(image_store)
    app = falcon.App()
    app.add_route('/images', image_resource)
    return app


def get_app():
    storage_path = os.environ.get('LOOK_STORAGE_PATH', '.')
    image_store = ImageStore(storage_path)
    return create_app(image_store)

现在,您可以针对所需的存储目录重新运行应用程序:

$ LOOK_STORAGE_PATH=/tmp gunicorn --reload 'look.app:get_app()'

现在您应该能够重新运行测试并看到它成功:

$ pytest tests -k test_posted_image_gets_saved

备注

上面的启动、测试、停止和每次测试运行后清理的过程都可以(而且确实应该)自动化。根据您的需要,您可以开发自己的自动化设备,或者使用库,例如 mountepy .

许多开发人员选择编写像上面这样的测试来健全地检查应用程序的主要功能,同时将大部分测试留给模拟请求和单元测试。与更高级的功能测试和系统测试相比,后一种类型的测试通常执行得更快,并且更容易实现更细粒度的测试断言。也就是说,测试策略差异很大,您应该选择最适合您需要的策略。

此时,您应该能够很好地掌握如何将通用测试策略应用到Falcon应用程序中。为了简洁起见,我们将省略以下章节中的进一步测试说明,重点是展示Falcon的更多特性。

服务图像

既然我们已经有了一种将图像输入服务的方法,我们当然需要一种将它们重新输出的方法。我们要做的是在请求图像时,使用位置头中返回的路径返回图像。

尝试执行以下操作:

$ http localhost:8000/images/db79e518-c8d3-4a87-93fe-38b620f9d410.png

作为回应,你应该得到 404 Not Found .这是falcon在找不到与请求的URL路径匹配的资源时给出的默认响应。

让我们通过创建一个单独的类来表示单个图像资源来解决这个问题。然后我们将添加一个 on_get() 方法来响应上面的路径。

继续编辑您的 images.py 文件的外观如下:

import io
import os
import re
import uuid
import mimetypes

import falcon
import msgpack


class Collection:

    def __init__(self, image_store):
        self._image_store = image_store

    def on_get(self, req, resp):
        # TODO: Modify this to return a list of href's based on
        # what images are actually available.
        doc = {
            'images': [
                {
                    'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
                }
            ]
        }

        resp.data = msgpack.packb(doc, use_bin_type=True)
        resp.content_type = falcon.MEDIA_MSGPACK
        resp.status = falcon.HTTP_200

    def on_post(self, req, resp):
        name = self._image_store.save(req.stream, req.content_type)
        resp.status = falcon.HTTP_201
        resp.location = '/images/' + name


class Item:

    def __init__(self, image_store):
        self._image_store = image_store

    def on_get(self, req, resp, name):
        resp.content_type = mimetypes.guess_type(name)[0]
        resp.stream, resp.content_length = self._image_store.open(name)


class ImageStore:

    _CHUNK_SIZE_BYTES = 4096
    _IMAGE_NAME_PATTERN = re.compile(
        '[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.[a-z]{2,4}$'
    )

    def __init__(self, storage_path, uuidgen=uuid.uuid4, fopen=io.open):
        self._storage_path = storage_path
        self._uuidgen = uuidgen
        self._fopen = fopen

    def save(self, image_stream, image_content_type):
        ext = mimetypes.guess_extension(image_content_type)
        name = '{uuid}{ext}'.format(uuid=self._uuidgen(), ext=ext)
        image_path = os.path.join(self._storage_path, name)

        with self._fopen(image_path, 'wb') as image_file:
            while True:
                chunk = image_stream.read(self._CHUNK_SIZE_BYTES)
                if not chunk:
                    break

                image_file.write(chunk)

        return name

    def open(self, name):
        # Always validate untrusted input!
        if not self._IMAGE_NAME_PATTERN.match(name):
            raise IOError('File not found')

        image_path = os.path.join(self._storage_path, name)
        stream = self._fopen(image_path, 'rb')
        content_length = os.path.getsize(image_path)

        return stream, content_length

如你所见,我们重新命名了 ResourceCollection 并添加了一个新的 Item 类来表示单个图像资源。或者,这两个类可以通过使用后缀响应器合并为一个类。(另见: add_route()

另外,请注意 name 的参数 on_get() 响应者。您在路由中指定的任何URI参数都将转换为相应的Kwarg并传递给目标响应程序。稍后我们将了解如何指定URI参数。

on_get() 响应,我们根据文件扩展名设置Content-Type头,然后直接从打开的文件句柄流出图像。请注意 resp.content_length 。无论何时使用 resp.stream 而不是 resp.textresp.data ,您通常还使用Content-Length标头指定流的预期长度,以便Web客户端知道要从响应中读取多少数据。

备注

如果您事先不知道流的大小,可以使用分块编码来解决这个问题,但这超出了本教程的范围。

如果 resp.status 未显式设置,默认为 200 OK ,这正是我们想要的 on_get() 去做。

现在让我们把所有的东西都连接起来试试。编辑 app.py 如下所示:

import os

import falcon

from .images import Collection, ImageStore, Item


def create_app(image_store):
    app = falcon.App()
    app.add_route('/images', Collection(image_store))
    app.add_route('/images/{name}', Item(image_store))
    return app


def get_app():
    storage_path = os.environ.get('LOOK_STORAGE_PATH', '.')
    image_store = ImageStore(storage_path)
    return create_app(image_store)

如你所见,我们指定了一条新路线, /images/{{name}} .这使得Falcon期望所有相关的响应者接受 name 争论。

备注

Falcon还支持包含多个值的更复杂的参数化路径段。例如,版本控制API可能使用以下路由模板来区分两个代码分支:

/repos/{org}/{repo}/compare/{usr0}:{branch0}...{usr1}:{branch1}

现在重新运行应用程序并尝试发布另一张图片:

$ http POST localhost:8000/images Content-Type:image/png < test.png

记下位置标题中返回的路径,然后使用它获取图像:

$ http localhost:8000/images/dddff30e-d2a6-4b57-be6a-b985ee67fa87.png

httpie不会显示图像,但您可以看到响应头设置正确。为了好玩,继续把上面的URI粘贴到你的浏览器中。图像应正确显示。

正在检查应用程序返回:

falcon-inspect-app look.app:get_app
Falcon App (WSGI)
• Routes:
    ⇒ /images - Collection:
       ├── GET - on_get
       └── POST - on_post
    ⇒ /images/{name} - Item:
       └── GET - on_get

介绍钩子

此时,您应该对组成基于falcon的API的基本部分有了相当好的理解。在我们完成之前,让我们花几分钟时间清理代码并添加一些错误处理。

首先,让我们在发布内容时检查传入的媒体类型,以确保它是常见的图像类型。我们将使用 before 钩子。

首先定义服务将接受的媒体类型列表。将此常量放在靠近顶部的位置,就在import语句之后 images.py

ALLOWED_IMAGE_TYPES = (
    'image/gif',
    'image/jpeg',
    'image/png',
)

这里的想法是只接受GIF、JPEG和PNG图像。如果您愿意,可以将其他人添加到列表中。

接下来,让我们创建一个钩子,它将在每个发送消息的请求之前运行。将此方法添加到以下定义中 ALLOWED_IMAGE_TYPES

def validate_image_type(req, resp, resource, params):
    if req.content_type not in ALLOWED_IMAGE_TYPES:
        msg = 'Image type not allowed. Must be PNG, JPEG, or GIF'
        raise falcon.HTTPBadRequest(title='Bad request', description=msg)

然后把钩子挂在 on_post() 响应者:

@falcon.before(validate_image_type)
def on_post(self, req, resp):
    pass

现在,在每次呼叫响应者之前,Falcon将首先调用 validate_image_type() .这个函数没有什么特别的地方,除了它必须接受四个参数。每个钩子的前两个参数都引用同一个钩子 reqresp 传递到响应程序的对象。这个 resource 参数是与请求关联的资源实例。第四个论点 params 按照惯例,这是一个参考Kwarg字典falcon为每个请求创建的。 params 将包含路由的URI模板参数及其值(如果有)。

正如您在上面的示例中看到的,您可以使用 req 获取有关传入请求的信息。但是,您也可以使用 resp 要根据需要使用HTTP响应,甚至可以使用钩子注入额外的Kwarg:

def extract_project_id(req, resp, resource, params):
    """Adds `project_id` to the list of params for all responders.

    Meant to be used as a `before` hook.
    """
    params['project_id'] = req.get_header('X-PROJECT-ID')

现在,您可以想象这样一个钩子应该应用于资源的所有响应者。实际上,通过简单地修饰类,钩子可以应用于整个资源:

@falcon.before(extract_project_id)
class Message:
    pass

类似的逻辑可以在中间件中全局应用。(另见: falcon.middleware

现在您已经添加了一个钩子来验证媒体类型,您可以通过尝试发布一些邪恶的内容来看到它的运行:

$ http POST localhost:8000/images Content-Type:image/jpx

你应该回去 400 Bad Request 状态和结构良好的错误体。

小技巧

当出现问题时,您通常希望向用户提供一些信息以帮助他们解决问题。此规则的例外情况是,由于用户被请求某些他们无权访问的内容而发生错误。在这种情况下,您可能只希望返回 404 Not Found 如果一个恶意用户正在寻找帮助他们破解你的应用程序的信息,那么你就需要一个空的主体。

退房 hooks reference 学习更多。

错误处理

一般来说,Falcon假设资源响应者 (on_get()on_post() 在大多数情况下,会做正确的事情。换句话说,Falcon并不努力保护响应程序代码不受其影响。

这种方法减少了(通常)Falcon必须执行的外来检查的数量,使框架更高效。有鉴于此,基于Falcon编写高质量的API需要:

  1. 资源响应程序将响应变量设置为正常值。

  2. 验证不受信任的输入(即来自外部客户机或服务的输入)。

  3. 您的代码经过了良好的测试,代码覆盖率很高。

  4. 在每个响应程序中或通过全局错误处理挂钩对错误进行适当的预测、检测、记录和处理。

当涉及到错误处理时,始终可以使用 resp 对象。然而,Falcon通过提供 set of error classes 如果出了问题,你可以加薪。Falcon将转换 falcon.HTTPError 由响应程序、挂钩或中间件组件引发到适当的HTTP响应中。

你可以举出 falcon.HTTPError 直接使用,或使用 predefined errors 它的设计目的是为每个错误类型适当地设置响应头和主体。

小技巧

可以为任何类型注册错误处理程序,包括 HTTPError .这个特性为日志记录和处理由响应程序、钩子和中间件组件引发的异常提供了一个中心位置。

参见: add_error_handler() .

让我们看一个简单的例子来说明这是如何工作的。尝试从应用程序请求无效的映像名:

$ http localhost:8000/images/voltron.png

正如你所看到的,结果并不完美。要解决这个问题,我们需要添加一些异常处理。修改您的 Item 分类如下:

class Item:

    def __init__(self, image_store):
        self._image_store = image_store

    def on_get(self, req, resp, name):
        resp.content_type = mimetypes.guess_type(name)[0]

        try:
            resp.stream, resp.content_length = self._image_store.open(name)
        except IOError:
            # Normally you would also log the error.
            raise falcon.HTTPNotFound()

现在,让我们再次尝试该请求:

$ http localhost:8000/images/voltron.png

有关错误处理的其他信息,请参见 error handling reference .

现在怎么办?

我们友好的社区可以回答您的问题,帮助您解决棘手的问题。另请参见: Getting Help .

如前所述,falcon的docstrings非常广泛,因此您只需从python repl中搜索falcon的模块就可以学到很多东西,例如 IPythonbpython .

此外,不要羞于在Github或您最喜欢的文本编辑器中提取Falcon的源代码。该团队试图使代码尽可能简单易读;如果其他文档可能不足,代码基本上不会出错。

在您的项目中可以使用许多Falcon附加组件、模板和补充包。我们已经在 Falcon wiki 作为一个起点,您可能还希望搜索pypi以获取其他资源。