Учебник

В этом учебнике мы пройдемся по созданию API для простого сервиса обмена изображениями. По пути мы обсудим основные возможности Falcon и представим терминологию, используемую фреймворком.

Первые шаги

Первое, что мы сделаем, это установка Falcon внутри свежего virtualenv. Для этого создадим новую папку проекта «look» и настроим виртуальную среду внутри неё, которую мы будем использовать для учебника:

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

Принято, чтобы модуль верхнего уровня проекта назывался так же, как и проект, поэтому создадим ещё одну папку «look» внутри первой и отметим её как модуль Python, создав пустой __init__.py файл внутри неё:

$ mkdir look
$ touch look/__init__.py

Далее, создадим новый файл, который будет точкой входа в ваше приложение:

$ touch look/app.py

Иерархия файлов должна теперь выглядеть так:

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

Теперь откройте app.py в вашем любимом текстовом редакторе и добавьте следующие строки:

import falcon

api = application = falcon.API()

Этот код создает ваше приложение WSGI и дает ему псевдоним api. Вы можете использовать любые имена переменных, но мы будем использовать application, так как это то, что Gunicorn по умолчанию ожидает (мы увидим, как это работает в следующей части учебника).

Далее посмотрим на класс falcon.API. Установите IPython и запустите его:

$ pip install ipython
$ ipython

Теперь введите следующее, чтобы получить информацию о вызываемом объекте falcon.API:

In [1]: import falcon

In [2]: falcon.API.__call__?

В качестве альтернативы вы можете использовать стандартную Python help() функцию:

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

Обратите внимание на сигнатуру метода. env и start_response — стандартные параметры WSGI. Falcon добавляет тонкий абстрактный слой поверх этих параметров, чтобы вам не приходилось взаимодействовать с ними напрямую.

Фреймворк Falcon содержит обширную встроенную документацию, которую вы можете запросить с помощью вышеуказанной техники.

Размещение приложения

Теперь, когда у вас есть простое приложение Falcon, вы можете запустить его с помощью WSGI-сервера. Python включает в себя сервер для самохостинга, но давайте воспользуемся чем-то более надёжным, что вы могли бы использовать в производстве.

Откройте новый терминал и выполните следующее:

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

(Обратите внимание на использование --reload опции, чтобы сказать Gunicorn перезагружать приложение всякий раз, когда меняется его код.)

Если вы пользователь Windows, вместо Gunicorn можно использовать Waitress, поскольку последний не работает под Windows:

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

Теперь, в другом терминале, попробуйте запросить работающее приложение с помощью curl:

$ curl -v localhost:8000

Вы должны получить 404. Это нормально, потому что мы ещё не указали никаких маршрутов. Falcon включает в себя обработчик ответа по умолчанию 404, который будет срабатывать для любого запрошенного пути, для которого маршрут не существует.

Хотя curl вполне справляется с задачей, его использование может быть немного громоздким. HTTPie — современная, удобная альтернатива. Давайте установим HTTPie и будем использовать его отныне:

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

Создание ресурсов

Архитектура Falcon заимствует несколько ключевых концепций из архитектурного стиля REST.

Ключевой концепцией и в REST, и в фреймворке 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 и т. д. Клиенты могут запросить ресурс, чтобы узнать, какие методы он поддерживает.

В зависимости от запрошенного действия сервер может или не может вернуть представление клиенту. Представления могут быть закодированы в любом из множества типов данных Интернета, таких как 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(object):

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

        # Create a JSON representation of the resource
        resp.body = 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 использует «утиную типизацию», поэтому вам не нужно наследоваться от какого-либо специального базового класса.

Ресурс изображения выше определяет единственный метод on_get(). Для любого HTTP-метода, который вы хотите, чтобы ваш ресурс поддерживал, просто добавьте метод on_*(), где * — любой стандартный HTTP-метод в нижнем регистре (например, on_get(), on_put(), on_head(), и т. д.).

Мы называем эти известные методы «ответчиками». Каждый ответчик принимает (по крайней мере) два параметра, один представляющий HTTP-запрос, а другой — HTTP-ответ на этот запрос. По соглашению они называются req и resp соответственно. Шаблоны маршрутов и хуки могут вводить дополнительные параметры, как мы увидим позже.

Сейчас ресурс изображений отвечает на запросы GET простым 200 OK и JSON-телом. По умолчанию тип данных Интернета в Falcon — application/json, но вы можете установить любой другой. Заслуживающие внимания альтернативы JSON включают YAML и MessagePack.

Теперь подключим этот ресурс и посмотрим, как это работает. Вернитесь к app.py и измените его так, чтобы он выглядел примерно так:

import falcon

from .images import Resource


api = application = falcon.API()

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

Теперь, когда поступает запрос на /images, Falcon будет вызывать ответчика на ресурсе изображений, соответствующем запрошенному HTTP-методу.

Давайте попробуем. Перезапустите Gunicorn (если вы не используете --reload) и отправьте GET-запрос на ресурс:

$ http localhost:8000/images

Вы должны получить 200 OK ответ, включая закодированное в JSON представление ресурса «изображения».

До сих пор мы реализовали только ответчик для GET. Давайте посмотрим, что произойдёт, когда будет запрошен другой метод:

$ http PUT localhost:8000/images

В этот раз вы получите 405 Method Not Allowed, так как ресурс не поддерживает метод PUT. Обратите внимание на значение заголовка Allow:

allow: GET, OPTIONS

Он генерируется автоматически Falcon на основе набора методов, реализованных целевым ресурсом. Если ресурс не включает собственный ответчик OPTIONS, фреймворк предоставляет реализацию по умолчанию. Поэтому OPTIONS всегда включены в список разрешённых методов.

Теперь, просто для интереса, давайте изменим наш ресурс, чтобы использовать MessagePack вместо JSON. Начните с установки соответствующего пакета:

$ pip install msgpack-python

Затем обновите ответчик, чтобы использовать новый тип данных:

import falcon

import msgpack


class Resource(object):

    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.body. Если вы присваиваете строку байтов последнему, Falcon разберётся, но вы можете получить небольшой прирост производительности, присвоив непосредственно resp.data.

Также обратите внимание на использование falcon.MEDIA_MSGPACK. Модуль falcon предоставляет ряд констант для распространённых типов данных, включая falcon.MEDIA_JSON, falcon.MEDIA_MSGPACK, falcon.MEDIA_YAML, falcon.MEDIA_XML, falcon.MEDIA_HTML, falcon.MEDIA_JS, falcon.MEDIA_TEXT, falcon.MEDIA_JPEG, falcon.MEDIA_PNG, и falcon.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 поддерживает тестирование своего объекта API, имитируя HTTP-запросы.

Тесты могут быть написаны с использованием стандартного модуля Python unittest или с использованием любого из ряда сторонних фреймворков тестирования, таких как pytest. В этом руководстве мы будем использовать pytest, так как он позволяет писать более питоновский код тестов по сравнению с модулем unittest в стиле JUnit.

Давайте начнем с установки пакета pytest:

$ pip install pytest

Далее, отредактируйте test_app.py так, чтобы он выглядел следующим образом:

import falcon
from falcon import testing
import msgpack
import pytest

from look.app import api


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


# 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, encoding='utf-8')

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

Из основного каталога проекта запустите свои новые тесты, выполнив pytest для каталога tests.

$ pytest tests

Если pytest сообщит об ошибках, потратьте немного времени, чтобы их исправить, прежде чем переходить к следующему разделу руководства.

Объекты запроса и ответа

Каждый обработчик в ресурсе получает объект Request, который можно использовать для чтения заголовков, параметров запроса и тела запроса. Вы можете использовать стандартную функцию help() или магическую функцию IPython ?, чтобы перечислить атрибуты и методы класса Falcon Request.

In [1]: import falcon

In [2]: falcon.Request?

Каждый обработчик также получает объект Response, который можно использовать для установки кода состояния, заголовков и тела ответа:

In [3]: falcon.Response?

Это будет полезно при создании конечной точки POST в приложении, которая может добавлять новые ресурсы изображений в нашу коллекцию. Мы займемся этой функциональностью далее.

Мы воспользуемся подходом TDD на этот раз, чтобы продемонстрировать, как применить эту стратегию тестирования при разработке приложения Falcon. С помощью тестов мы сначала точно определим, что должно делать приложение, а затем будем писать код до тех пор, пока тесты не скажут нам, что мы закончили.

Давайте начнём с добавления дополнительного оператора импорта в test_app.py. Нам нужно импортировать два модуля из unittest.mock, если вы используете Python 3, или из mock, если вы используете Python 2.

# Python 3
from unittest.mock import mock_open, call

# Python 2
from mock import mock_open, call

Для Python 2 вам также потребуется установить пакет mock.

$ pip install mock

Теперь добавьте следующий тест:

# "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

Чтобы заставить новый тест пройти, нам нужно добавить новый метод для обработки POST-запросов. Откройте images.py и добавьте обработчик POST в класс Resource следующим образом:

import io
import os
import uuid
import mimetypes

import falcon
import msgpack


class Resource(object):

    _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() устанавливает заголовок Location для вновь созданного ресурса. (Мы создадим маршрут для этого пути через минуту.) Классы Request и Response содержат удобные атрибуты для чтения и установки общих заголовков, но вы всегда можете получить доступ к любому заголовку по имени с помощью методов 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

Теперь, если вы проверите свой каталог хранения, он должен содержать копию изображения, которое вы только что отправили.

Вперед!

Рефакторинг для обеспечения тестируемости

Ранее мы отметили, что наш тест POST сильно опирался на подделку, полагаясь на предположения, которые могут или не могут быть верными по мере развития кода. Чтобы смягчить эту проблему, нам нужно не только переработать тесты, но и само приложение.

Мы начнём с выделения бизнес-логики из обработчика POST ресурса в images.py, чтобы ее можно было тестировать независимо. В данном случае «бизнес-логикой» ресурса является просто операция сохранения изображения:

import io
import mimetypes
import os
import uuid

import falcon
import msgpack


class Resource(object):

    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(object):

    _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


api = application = falcon.API()

image_store = ImageStore('.')
images = Resource(image_store)
api.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)
    api = falcon.API()
    api.add_route('/images', image_resource)
    return api


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

Как вы можете видеть, основная логика настройки была перемещена в create_app(), которая может использоваться для получения объекта API для тестирования или размещения в производстве. get_app() заботится о создании дополнительных ресурсов и настройке приложения для размещения.

Команда для запуска приложения теперь:

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

Наконец, нам нужно обновить код теста. Измените test_app.py следующим образом:

import io

# Python 3
from unittest.mock import call, MagicMock, mock_open

# Python 2
# from 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):
    api = look.app.create_app(mock_store)
    return testing.TestClient(api)


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, encoding='utf-8')

    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], falcon.request_helpers.BoundedStream)
    assert saver_call[0][1] == image_content_type

Как видите, мы переделали POST. Хотя подделок меньше, утверждения стали более подробными, чтобы правильно проверить взаимодействия на границах интерфейса.

Проверим наш прогресс:

$ 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

Как и предыдущий тест, этот всё ещё использует подделки. Но структура кода была улучшена путём компонентного проектирования и инверсии зависимостей, что делает приложение более гибким и тестируемым.

Функциональные тесты

Функциональные тесты определяют поведение приложения извне. При использовании TDD это может быть более естественным началом по сравнению с тестированием на уровне единиц, поскольку сложно заранее предвидеть, какие внутренние интерфейсы и компоненты необходимы до определения пользовательского интерфейса приложения.

В случае рефакторинга из предыдущего раздела мы могли непреднамеренно ввести функциональную ошибку в приложение, которую наши тесты на уровне единиц не обнаружили бы. Это может произойти, когда ошибка является результатом непредвиденного взаимодействия между несколькими блоками, между приложением и веб-сервером или между приложением и любыми внешними сервисами, от которых оно зависит.

С помощью таких вспомогательных функций тестирования, как 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)
    api = falcon.API()
    api.add_route('/images', image_resource)
    return api


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

Многие разработчики выбирают писать тесты, подобные приведенному выше, для проверки работоспособности основных функций своего приложения, оставляя основную часть тестирования симулированным запросам и модульным тестам. Эти последние типы тестов, как правило, выполняются гораздо быстрее и позволяют проводить более точные утверждения в тестах по сравнению с функциональными и системными тестами высокого уровня. При этом стратегии тестирования сильно различаются, и вы должны выбрать стратегию, которая лучше всего подходит для ваших потребностей.

На данный момент у вас должно быть хорошее понимание того, как применять распространенные стратегии тестирования к вашему приложению Falcon. В целях краткости мы опустим дальнейшие инструкции по тестированию из следующих разделов, сосредоточившись вместо этого на демонстрации большего количества функций Falcon.

Отображение изображений

Теперь, когда у нас есть способ размещения изображений в службе, конечно, нам нужен способ их извлечения. Мы хотим возвращать изображение при запросе, используя путь, возвращенный в заголовке Location.

Попробуйте выполнить следующее:

$ 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(object):

    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(object):

    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.stream_len = self._image_store.open(name)


class ImageStore(object):

    _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')
        stream_len = os.path.getsize(image_path)

        return stream, stream_len

Как видите, мы переименовали Resource в Collection и добавили новый класс Item для представления одного ресурса изображения. Также обратите внимание на параметр name для ответчика on_get(). Любые параметры URI, которые вы укажете в своих маршрутах, будут преобразованы в соответствующие значения kwargs и переданы целевому ответчику как таковые. Мы увидим, как указать параметры URI чуть позже.

Внутри ответчика on_get() мы устанавливаем заголовок Content-Type на основе расширения файла, а затем передаем изображение напрямую из открытого файла. Обратите внимание на использование resp.stream_len. При использовании resp.stream вместо resp.body или resp.data, вы обычно также указываете ожидаемую длину потока, чтобы веб-клиент знал, сколько данных следует читать из ответа.

Если resp.status не задано явно, оно по умолчанию равно 200 OK, что именно то, что мы хотим, чтобы on_get() сделал.

Теперь давайте настроим все и попробуем. Откройте файл app.py и измените его на что-то похожее на следующее:

import os

import falcon

import images


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


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

Как видите, мы указали новый маршрут /images/{name}. Это заставляет Falcon ожидать, что все связанные ответчики будут принимать аргумент name.

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

Теперь перезапустите приложение и попробуйте отправить еще одно изображение:

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

Запишите путь, возвращённый в заголовке Location, и используйте его для получения изображения:

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

HTTPie не отобразит изображение, но вы увидите, что заголовки ответа были установлены правильно. Просто для интереса скопируйте указанный выше URI в свой браузер. Изображение должно отображаться правильно.

Введение в хуки

На данном этапе вы должны хорошо понимать основные части, составляющие API на основе Falcon. Перед завершением давайте потратим немного времени на упорядочивание кода и добавление обработки ошибок.

Сначала давайте проверим входящий тип медиа при отправке чего-либо, чтобы убедиться, что это распространенный тип изображения. Мы реализуем это с помощью хука before.

Начните с определения списка типов медиа, которые служба будет принимать. Разместите эту константу вверху, сразу после операторов импорта в файле 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('Bad request', msg)

А затем подключим хук к ответчику on_post():

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

Теперь перед каждым вызовом этого ответчика Falcon сначала вызовет validate_image_type(). В этой функции нет ничего особенного, кроме того, что она должна принимать четыре аргумента. Каждый хук принимает в качестве своих первых двух аргументов ссылку на те же объекты req и resp, которые передаются в ответчики. Аргумент resource — это экземпляр Resource, связанный с запросом. Четвёртый аргумент, по соглашению названный params, является ссылкой на словарь kwargs, который Falcon создаёт для каждого запроса. params будет содержать параметры шаблона URI маршрута и их значения, если таковые имеются.

Как видно из примера выше, вы можете использовать req для получения информации о входящем запросе. Однако вы также можете использовать resp для работы с HTTP-ответом по мере необходимости, и вы даже можете использовать хуки для вставки дополнительных значений kwargs.

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(object):

    # ...

Аналогичная логика может быть применена глобально с помощью middleware. (См. также: falcon.middleware)

Теперь, когда вы добавили хук для проверки типа медиа, вы можете увидеть его в действии, попытавшись отправить что-то нежелательное:

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

Вы должны получить код состояния 400 Bad Request и хорошо структурированное тело ошибки.

Посмотрите справочник по хукам, чтобы узнать больше.

Обработка ошибок

В общем случае Falcon предполагает, что ответчики ресурса (on_get(), on_post(), и т. д.) в основном будут действовать правильно. Другими словами, Falcon не старается защитить код ответчика от самого себя.

Этот подход уменьшает количество (часто) лишних проверок, которые Falcon в противном случае должен был бы выполнять, делая фреймворк более эффективным. Имея это в виду, для написания качественного API на основе Falcon требуется:

1. Ответчики ресурсов устанавливают переменные ответа в разумные значения.
2. Ненадёжные данные (т. е. данные от внешнего клиента или службы) валидируются.
3. Код хорошо протестирован с высокой степенью покрытия кода.
4. Ошибки предвидены, обнаружены, зарегистрированы и обработаны надлежащим образом внутри каждого ответчика или с помощью глобальных хуков обработки ошибок.

Что касается обработки ошибок, вы всегда можете напрямую установить код состояния ошибки, соответствующие заголовки ответа и тело ошибки с помощью объекта resp. Однако Falcon пытается сделать всё проще, предоставив набор классов ошибок, которые вы можете генерировать при возникновении проблемы. Falcon преобразует любой экземпляр или подкласс falcon.HTTPError, возбужденный ответчиком, хуком или компонентом middleware, в соответствующий HTTP-ответ.

Вы можете непосредственно генерировать экземпляр falcon.HTTPError или использовать любой из предопределённых ошибок, предназначенных для корректной настройки заголовков и тела ответа для каждого типа ошибки.

Давайте посмотрим быстрый пример, как это работает. Попробуйте запросить недопустимое имя изображения у вашего приложения:

$ http localhost:8000/images/voltron.png

Как вы видите, результат не очень приятный. Для решения этой проблемы нам нужно добавить некоторую обработку исключений. Измените класс Item следующим образом:

class Item(object):

    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.stream_len = self._image_store.open(name)
        except IOError:
            # Normally you would also log the error.
            raise falcon.HTTPNotFound()

Теперь давайте попробуем этот запрос снова:

$ http localhost:8000/images/voltron.png

Дополнительная информация об обработке ошибок доступна в справочнике по обработке ошибок.

Что дальше?

Наша дружелюбная сообщество готова ответить на ваши вопросы и помочь вам решить сложные задачи. См. также: Получение помощи.

Как упоминалось ранее, документация Falcon довольно обширная, и вы можете многому научиться, просто поэкспериментировав с модулями Falcon из интерактивной оболочки Python, такой как IPython или bpython.

Также не стесняйтесь просматривать исходный код Falcon на GitHub или в вашем любимом текстовом редакторе. Команда постаралась сделать код как можно более простым и читаемым; там, где может не хватить другой документации, код вряд ли будет ошибочным.

Набор дополнений, шаблонов и дополнительных пакетов для использования в ваших проектах доступны для использования с Falcon. Некоторые из них перечислены на сайте Falcon wiki в качестве отправной точки, но вы также можете найти дополнительные ресурсы в PyPI.