Рубрики
Без рубрики

Быстрый и грязный макет с Starlette

У меня была проблема на работе. Команде нужно было издеваться над сторонним обслуживанием в среде тестирования … Tagged с Python, Testing, WebDev, Starlette.

У меня была проблема на работе. Команде нужно было издеваться над сторонним обслуживанием в среде тестирования. Служба была медленной, и настройка была болезненной. Если бы мы могли издеваться над этим, то команда могла бы избежать этих проблем.

Проблема с высмеиванием сервиса заключается в том, что часть потока должна вызвать веб -крючок, который перезвонит в систему моей компании, чтобы указать, что вся работа выполнена. Кроме того, вызов Webhook должен быть после задержки, потому что услуга, которую мы моделируем, занимает много времени (то есть более 60 секунд).

Проблема: Запустите высмеивающуюся стороннюю службу с задержкой обратного вызова Webhook.

Решение должно:

  • Ответьте на запрос сообщения и вернуть 200 ok статус.
  • Позвоните в Webhook обратно в систему, которая будет включать дополнительные идентификаторы для подключения вызова Webhook к исходному запросу.
  • Будьте как можно проще, потому что это не основной продукт.

Параметры решения

Экосистема Python полна сильных вариантов для решения первой части решения. Джанго, колба, пирамида, бутылка и любая другая веб -структура, о которой вы можете подумать, с легкостью справится с этим.

Вторая часть решения сложнее. Если я не осторожен, то простое выходу из окна, и я уничтожил третью цель.

Фоновые инструменты задачи

Как бы я ни люблю Джанго, я не думал, что это было хорошо. В моем мыслительном процессе вызов WebHook позже потребовал какую -то фоновую задачу. Если вы знаете Джанго, то вы можете сразу подумать о Сельдерей , и это была моя первая мысль. Проблема в том, что сельдерею требуется брокер сообщения, как Rabbitmq . При этом бюджет простоты полностью снят.

Альтернатива сельдерею – это RQ , но у него есть похожие проблемы и требуют запуска Redis наряду с отдельным рабочим процессом.

Можем ли мы решить эту проблему без использования отдельного инструмента фоновой задачи?

Без справочных инструментов задач

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

Я рассмотрел два подхода:

  1. Используйте потоки.
  2. Используйте совместное параллелизм (прочитайте: асинхронное программирование)

Я не люблю резьбое программирование в Python. На самом деле, мне не нравится программирование с резьбой. 1 Решение с потоком может полностью работать, выполняя вызов WebHook в отдельном потоке, но меня интересовали другие варианты.

Могу ли я решить проблему с асинхронным программированием? Асинхронное программирование намного проще в недавних версиях Python. На языке теперь есть встроенные ключевые слова, такие как Асинхронизация и ждет Это делает программирование в этом стиле более дружелюбным. Я оценил несколько Async Web Frameworks, и я вспомнил функцию, которую я видел в Starlette Документация: Фоновые задачи

Starlette – это веб -структура, разработанная автором Django Rest Framework (DRF) , Том Кристи Анкет DRF – такой солидный проект. Разделение того же создателя укрепил мою уверенность в том, что Starlette будет хорошо разработанным программным обеспечением.

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

В остальной части этой статьи будет рассмотрено, что я сделал, чтобы решить проблему, используя Starlette с общей версией, которая пропускает скучные детали третьей стороны.

Starlette

Поскольку Starlette – это асинхронная структура, инструменты Python, необходимые для поддержки приложения Starlette, отличаются от стандартного Wsgi приложение.

Вместо синхронно Веб -сервер как Онломщик или Uwsgi , а асинхронно Требуется веб -сервер. Я выбрал Uvicorn , веб -сервер, похожий на стреляющегося, который использует uvloop Чтобы справиться с асинхронной петлей события.

Чтобы сделать обратный вызов Webhook, эмулятор должен выполнять HTTP -запросы. Натуральный пакет для достижения Запросы , но Запросы является синхронным пакетом, который заблокирует петлю события. Мне нужна была асинхронная библиотека HTTP -запросов. Для этого требования я выбрал Httpx Анкет HTTPX – это очень новая библиотека, которая разделяет тот же API, что и Запросы и Работает с асинхронным программированием.

Если вы хотите следовать, сначала установите все необходимые зависимости. Вам понадобится хотя бы Python 3.6.

$ mkdir mock-service
$ cd mock-service
$ python3 -m venv venv
$ . venv/bin/activate
(venv) $ pip install starlette uvicorn httpx

Для будущих читателей, вот версии, которые PIP установка Команда установлена в мою локальную виртуальную среду на момент написания этой статьи.

$ pip freeze
certifi==2019.6.16
chardet==3.0.4
Click==7.0
h11==0.8.1
h2==3.1.1
hpack==3.0.0
hstspreload==2019.8.20
httptools==0.0.13
httpx==0.7.1
hyperframe==5.2.0
idna==2.8
rfc3986==1.3.2
starlette==0.12.8
uvicorn==0.8.6
uvloop==0.12.2
websockets==7.0

Давайте начнем с перечисления полного примера. Я сломаю каждую часть приложения, но иногда это помогает сначала получить полную картину. Чтобы попробовать это самостоятельно, вы можете захватить Пример скрипта Анкет

import asyncio
import os
import uuid

import httpx
from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.responses import JSONResponse
import uvicorn

app = Starlette()
client = httpx.AsyncClient()
CALLBACK_URL = os.environ["CALLBACK_URL"]


@app.route("/api/endpoint", methods=["POST"])
async def fake_endpoint(request):
    identifier = str(uuid.uuid4())
    payload = {
        "identifier": identifier,
        "some_parameter": request.query_params.get("some_parameter"),
    }
    task = BackgroundTask(trigger_webhook, payload)
    return JSONResponse(
        {"identifier": identifier, "success": True}, background=task)


async def trigger_webhook(payload):
    await asyncio.sleep(5)
    params = {
        "success": True,
        "identifier": payload["identifier"],
        "some_parameter": payload["some_parameter"],
    }
    await client.get(CALLBACK_URL, params=params)


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Запуск кода

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

Во -первых, нам нужен приемник Webhook. Встроенный HTTP-сервер Python идеально подходит для этой задачи. В отдельном терминале запустите:

$ python3 -m http.server 5000
Serving HTTP on 0.0.0.0 port 5000 (http://0.0.0.0:5000/) ...

Далее, запустите фиктивную службу.

(venv) $ CALLBACK_URL=http://0.0.0.0:5000 python3 mock_service.py
INFO: Started server process [47148]
INFO: Waiting for application startup.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

Заметьте, что я прошел в Callback_url как переменная среды с URL -адресом от нашего приемника WebHook. Служба Mock будет использовать эту информацию, чтобы узнать, куда отправить запрос Webhook.

С основными кусочками нам нужно запустить макет обслуживания. Мне нравится Httpie Как более дружелюбная альтернатива Curl Анкет

(venv) $ pip install httpie

Теперь мы можем заставить все двигаться! Давайте выпустим сообщение в фиктивную службу.

(venv) $ http POST :8000/api/endpoint some_parameter==some_value
HTTP/1.1 200 OK
content-length: 68
content-type: application/json
date: Sat, 24 Aug 2019 15:54:07 GMT
server: uvicorn

{
    "identifier": "4a3a0ce2-ae0c-41d3-ba58-89a5a9579692",
    "success": true
}

Из Mock Service вы увидите журнал, как:

INFO: ('127.0.0.1', 61647) - \
    "POST /api/endpoint?some_parameter=some_value HTTP/1.1" 200

Пять секунд спустя, фальшивый приемник Webhook должен показать:

127.0.0.1 - - [24/Aug/2019 11:54:12] \
    "GET /?success=true&identifier= \
    4a3a0ce2-ae0c-41d3-ba58-89a5a9579692&some_parameter=some_value \
    HTTP/1.1" 200 -

Я переформатировал линии, чтобы они подойдут лучше в статье.

Потрясающий! Мик служба сделал именно то, что мы хотели. Запрос POST получил успешный ответ, и после задержки Webhook получил одну и ту же информацию.

Теперь, когда мы видели, как все работает , давайте разберем код.

Объяснение Mock Service

Мы можем посмотреть этот код в нескольких отдельных кусках.

import asyncio
import os
import uuid

import httpx
from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.responses import JSONResponse
import uvicorn

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

app = Starlette()
client = httpx.AsyncClient()
CALLBACK_URL = os.environ["CALLBACK_URL"]

Этот раздел создает глобальные уровни модуля, которые мы собираемся использовать. Иногда глобалы являются лучшим инструментом для работы, даже если они часто осуждаются. Callback_url это единственная настройка для эмулятора. Обратите внимание, что я получаю значение от Os.environ С индексом синтаксиса вместо os.environ.get ('callback_url') синтаксис. Индексный стиль извлечения обеспечит, чтобы у нас было значение, поскольку необходимо правильно управлять эмулятором.

@app.route("/api/endpoint", methods=["POST"])
async def fake_endpoint(request):
    identifier = str(uuid.uuid4())
    payload = {
        "identifier": identifier,
        "some_parameter": request.query_params.get("some_parameter"),
    }
    task = BackgroundTask(trigger_webhook, payload)
    return JSONResponse(
        {"identifier": identifier, "success": True}, background=task)

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

  1. Генерирует идентификатор, который звонящий мог бы использовать, чтобы связать этот звонок с веб -крюком, который будет следовать.
  2. Извлекает значение из строки запроса запроса, some_parameter и хранит его для фоновой задачи.
  3. Связочные данные в полезную нагрузку, которая будет доступна для Фоновая карта который выполнит trigger_webhook функция
  4. Отвечает идентификатором и устанавливает фоновую задачу для запуска.

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

async def trigger_webhook(payload):
    await asyncio.sleep(5)
    params = {
        "success": True,
        "identifier": payload["identifier"],
        "some_parameter": payload["some_parameter"],
    }
    await client.get(CALLBACK_URL, params=params)

Триггер Webhook спит в течение 5 секунд. Важно использовать asyncio.sleep вместо Time.sleep Анкет Функция сна от время является синхронной командой и заблокирует цикл события. Позвонив Asyncio.sleep с Ждите Функция дает выполнение обратно в цикл события до 5 секунд.

Как только задержка закончится, httpx Клиент Вызывает URL -адрес обратного вызова, чтобы вернуть идентификатор и параметр. Это указывает на вызывающего абонента, что эмулятор выполнен.

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Последний бит кода соединяет приложение к Uvicorn, чтобы Uvicorn запустился, когда сервис будет вызван:

(venv) $ python3 mock_service.py

Старлетт -много многообещающего будущего

API Starlette сделал этот эмулятор таким же быстрым производством, насколько я мог бы надеяться. Структура все еще нова, но я думаю, что у нее есть очень многообещающее будущее в экосистеме Python. У него уже есть наилучшая производительность любого питона в рамках Techempower Benchmarks Анкет

Я надеюсь, что вам понравилось смотреть на какое -то асинхронное программирование в Python. Если у вас есть какие -либо вопросы, пожалуйста, поделитесь в Twitter и свяжитесь со мной по адресу @mblayman Анкет

Спасибо за чтение!

Эта статья впервые появилась на mattlayman.com Анкет

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

Оригинал: “https://dev.to/mblayman/quick-and-dirty-mock-service-with-starlette-cfh”