Учебник

В этом учебнике мы пройдемся по созданию 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:api

Теперь, в другом терминале, попробуйте запросить работающее приложение с помощью 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». Создайте новый модуль 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-кодированное представление ресурса «images».

Пока что мы реализовали обработчик только для 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, raw=False)

    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. С помощью тестов мы сначала точно определим, что должно сделать приложение, а затем напишем код, пока тесты не скажут нам, что мы закончили.

Начнём с добавления дополнительной инструкции import в 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 Created». Мы также могли использовать псевдоним 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 тест сильно опирался на имитацию, полагаясь на предположения, которые могут или не могут быть верными по мере развития кода. Чтобы смягчить эту проблему, нам нужно не только переработать тесты, но и само приложение.

Начнём с вынесения бизнес-логики из 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, 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], 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.content_length = 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')
        content_length = os.path.getsize(image_path)

        return stream, content_length

Как видите, мы переименовали Resource в Collection и добавили новый класс Item для представления отдельного ресурса изображения. В качестве альтернативы, эти два класса можно объединить в один, используя суффиксные ответчики. (См. также: add_route())

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

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

Если 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.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

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

Что дальше?

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

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

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

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