Учебник

В этом руководстве мы пройдёмся по созданию 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 = 'application/msgpack'
        resp.status = falcon.HTTP_200

Обратите внимание на использование resp.data вместо resp.body. Если вы присваиваете строку байтов последнему, Falcon это поймёт, но вы можете получить небольшое повышение производительности, присвоив её напрямую resp.data.

Перезапустите 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 ? для перечисления атрибутов и методов класса Request Falcon:

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 = 'application/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 = 'application/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 = 'application/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, которые вы укажете в своих маршрутах, будут преобразованы в соответствующие ключевые аргументы и переданы целевому обработчику как таковые. Мы увидим, как указать параметры 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 — это экземпляр ресурса, связанный с запросом. Четвёртый аргумент, названный params по соглашению, — это ссылка на словарь ключевых аргументов, созданный Falcon для каждого запроса. params будет содержать параметры шаблона URI маршрута и их значения, если таковые имеются.

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

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

    # ...

Аналогичная логика может применяться глобально с помощью миддлверов. (См. также: 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, поднятый обработчиком, хуком или компонентом миддлвеара, в соответствующий 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 REPL, такого как IPython или bpython.

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

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