Автор оригинала: Oyeniyi Abiola.
Поскольку введение Asyncio в Python 3.5+, большое количество Async Web Frameworks в Python была на подъеме, некоторые из которых включают, но не ограничиваются
Основными преимуществами этих Async Web Framework являются то, что они могут справиться с большим количеством запросов и, как правило, быстрее, чем их регулярные аналоги синхронизации.
Что-то, с которым я боролся, было, как интегрировать использование любого из этих веб-каркасов в мое текущее существующее приложение, написанное в Джанго Так что я мог извлечь выгоду из этих улучшений скорости.
В то же время мне не интересовалось переписать каждый аспект моего приложения с нуля и по-прежнему заинтересованы в использовании Django ORM с любым решением, с которым я придумал, потому что это самый простой ORM, который я знаю в Python. Я был активно следовать разработке Каналы Django , попытка, которая берет Django и расширяет свои способности за пределами HTTP – для обработки WebSockets, протоколов чата, протоколы IoT и многое другое. Я был надежен и оптимистичен о том, какие возможности такое развитие приведут для разработчиков Django. Но я никогда не продал 100%, используя его для моих проектов в основном, потому что это требовало использование внешней зависимости на Redis и бежать в здание, скрученные на окна, во время разработки.
Я в конечном итоге узнал о Starlette , легкий ASGI Framework/Toolkit, который идеально подходит для строительства высокопроизводительных услуг Asyncio. Я узнал об этом из-за разработки Увикорн , молниеносный сервер ASGI. После прохождения документации я мог сразу увидеть, как Starlette может быть введен в мой текущий рабочий процесс без необходимости делать так много работы. Что сделано Starlette Выделись, на мой взгляд, был тот факт, что он был рекламирован как инструментарий. Это означало, что вы можете использовать некоторые из его функций в Существующий проект или как автономное веб-каркас. Проходя через документацию, функция, которая зажег для меня лампы, была поддержка Фоновые задачи Отказ Если вы когда-либо делали какую-либо форму обработки фона в Джанго Раньше вы знаете, что вам нужна сторонняя библиотека, как Сельдерей или RQ вместе с [RABBITMQ] [RABBITMQ] или Redis как сообщение брокера. Теперь я получаю эту же функциональность без дополнительных зависимостей.
NB: имейте в виду, что для длительных фоновых задач и периодических действий сельдерея могут быть лучше. В этом сценарии это было в основном извилистым
Следующим шагом было найти проблему, чтобы заставить мое использование Starlette Отказ Я столкнулся с сценарием, где я испытывал ошибки для QuickBooks API, когда пытаетесь создать квитанцию о продаже после оплаты. Поскольку это действие не было точно потребовалось до того, как ответ был отправлен обратно клиенту, это был хороший кандидат в качестве фоновой задачи.
Загородный код показан ниже
def verify_payment(request, order):
amount = request.GET.get("amount")
txrf = request.GET.get("trxref")
paystack_instance = PaystackAPI()
response = paystack_instance.verify_payment(txrf, amount=int(amount))
if response[0]:
p_signals.payment_verified.send(
sender=PaystackAPI, ref=txrf, amount=int(amount), order=order
)
return JsonResponse({"success": True})
return JsonResponse({"success": False}, status=400)
и обратный вызов сигнала был это
@receiver(p_signals.payment_verified)
def on_payment_verified(sender, ref, amount, order, **kwargs):
record = UserPayment.objects.filter(order=order).first()
record.made_payment = True
record.amount = Decimal(amount) / 100
record.save()
# process to quickbooks
record.create_sales_receipt()
record.add_to_mailing_list()
Способ представления в настоящее время структурирован до получения получения продаж. В идеале, как только платеж был проверен, остальное действие может быть запущено на заднем плане.
Я создал образец Starlette Вид для копирования функциональности, предоставляемых Django
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.exceptions import ExceptionMiddleware
from starlette.background import BackgroundTask
from starlette.middleware.cors import CORSMiddleware
app = Starlette()
app.add_middleware(CORSMiddleware, allow_origins=['*'])
...
@app.route("/paystack/verify-payment/{order}/")
async def paystack_verify_payment(request, order):
response = await services.verify_payment(request.query_params, order)
if response[0]:
task = BackgroundTask(process_payment, request.query_params, order)
return JSONResponse({"success": True}, background=task)
return JSONResponse({"success": False}, status_code=400)
Мой Services.py состоял из следующего.
from asgiref.sync import sync_to_async
import django
django.setup()
...
@sync_to_async
def verify_payment(request, order):
amount = request.get("amount")
txrf = request.get("trxref")
paystack_instance = PaystackAPI()
return paystack_instance.verify_payment(txrf, amount=int(amount))
Как я упоминал ранее, мне не интересовало необходимость изменить большую часть заявки. Библиотеки, как Asgiref много помогло при переводу синхронных функций в COROUTINE, который может быть ожидается. Также, так как я собирался использовать Джанго ORM за пределами Django I.E Starlette Мне нужно добавить django.setup () Перед тем, как сделать какой-либо конкретный вызов модели Django.
Process_Payment Фоновая задача также должна быть Функция Coroutine
async def process_payment(request, order):
services.process_paystack_payment(request, order)
Последний кусок головоломки обеспечивал, чтобы другие взгляды обслуживались Джанго Хотя этот конкретный вид был обслужен Starlette Отказ Копаться через Starlette Исходный код, я наткнулся на следующие промежуточные программы Starlette.middleware.wsgi.wsgimiddleware Отказ Я смог вывести из этого, что имело возможность обернуть приложение WSGI в качестве приложения ASGI. Поскольку Starlette может легко составить различные приложения ASGI как выделенные маршруты, я закончил с помощью следующего помощника функции
from starlette.applications import Starlette
from starlette.middleware.wsgi import WSGIMiddleware
from starlette.routing import Router, Path, PathPrefix
def create_asgi_app():
return Starlette()
def create_app(application, wsgi=False):
if wsgi:
return WSGIMiddleware(application)
return application
def initialize_router(apps):
return Router(
[PathPrefix(x["path"], app=create_app(x["app"], x.get("wsgi"))) for x in apps]
)
Наконец, я создал файл Python Point Point, который будет ссылаться на UVicorn, веб-сервер ASGI
from payment_service.wsgi import application
from v2 import app as asgi_app
from cv_utils.starlette import initialize_router
app = initialize_router(
[{"path": "/v2", "app": asgi_app}, {"path": "", "app": application, "wsgi": True}]
)
Экземпляр WSGI для приложения Django можно найти в wsgi.py файл
Приложение ASGI Starlette будет обработать любой запрос на /v2 Путь, в то время как все другие маршруты будут отправлены в приложение Django.
Так как это было сообщено в Starlette Документы, что боевика , тестированный битранский сервер приложений в Python может хорошо играть с UVicorn, я закончил изменять свою конфигурацию DockerFile до следующего
CMD gunicorn --workers=4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:5000 run:app --access-logfile -
Мне также нужно было обновить свой автоматизированный тест для этого конкретного представления.
Старая реализация выглядела так
...
@mock.patch("payment_service.models.m_send_mail")
@mock.patch("requests.post")
@mock.patch("payment_service.models.QuickbooksAPI")
@mock.patch("payment_service.urls.PaystackAPI")
def test_payment_made_with_pastack(
self, mock_paystack, mock_q_books, mock_post, mock_mail
):
mock_instance = self.get_mock(mock_paystack, (True, "verification successful"))
mock_quickbooks = mock_q_books.return_value
mock_quickbooks.create_customer.return_value = {
"id": "23",
"name": "Danny Novaka",
}
mock_quickbooks.create_sales_receipt.return_value = "2322"
...
with self.env:
response = self.client.get(
"/paystack/verify-payment/ADESFG123453/",
{"amount": 2000 * 100, "trxref": "freeze me"},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), {"success": True})
mock_instance.verify_payment.assert_called_once_with(
"freeze me", amount=200000
)
record = models.UserPayment.objects.first()
self.assertTrue(record.made_payment)
extra_data = record.extra_data
self.assertEqual(
extra_data["quickbooks_customer_details"],
{"id": "23", "name": "Danny Novaka"},
)
self.assertEqual(extra_data["quickbooks_receipt_id"], "2322")
mock_quickbooks.create_customer.assert_called_once_with(
**{
...
)
mock_quickbooks.create_sales_receipt.assert_called_once_with(
{"id": "23", "name": "Danny Novaka"},
{
"currency": record.currency,
"description": record.description,
"price": record.price,
"amount": record.price,
"discount": 0,
},
)
mock_post.assert_called_once_with(
settings.AUTH_ENDPOINT + "/save-custom-data",
json={
"user_id": record.user,
"data": {
"quickbooks_customer_details": {
"id": "23",
"name": "Danny Novaka",
}
},
},
)
mock_mail.assert_called_once_with(
"first_resume_download",
{"first_name": "Danny"},
[record.extra_data["email"]],
)
Из-за уровня издевательства, мне нужно было установить Pteest , Pytest-django и pytest-mocker Чтобы помочь мне в написании эквивалентного теста в Starlette
@pytest.mark.django_db
def test_post(mocker, monkeypatch):
mock_paystack = mocker.patch("payment_service.services.PaystackAPI")
mock_mail = mocker.patch("payment_service.models.m_send_mail")
mock_post = mocker.patch("requests.post")
mock_q_books = mocker.patch("payment_service.models.QuickbooksAPI")
monkeypatch.setenv("PAYSTACK_SECRET_KEY", "MY-SECRET-KEY")
mock_instance = get_mock(mock_paystack, (True, "verification successful"))
mock_quickbooks = mock_q_books.return_value
mock_quickbooks.create_customer.return_value = {"id": "23", "name": "Danny Novaka"}
mock_quickbooks.create_sales_receipt.return_value = "2322"
...
client = TestClient(app)
response = client.get(
"/v2/paystack/verify-payment/ADESFG123453/",
params={"amount": 2000 * 100, "trxref": "freeze me"},
)
assert response.status_code == 200
assert response.json() == {"success": True}
mock_instance.verify_payment.assert_called_once_with("freeze me", amount=200000)
record = models.UserPayment.objects.first()
assert record.made_payment == True
extra_data = record.extra_data
assert extra_data["quickbooks_customer_details"] == {
"id": "23",
"name": "Danny Novaka",
}
assert extra_data["quickbooks_receipt_id"] == "2322"
mock_quickbooks.create_customer.assert_called_once_with(
**{
"email": record.extra_data["email"],
"full_name": f"{record.extra_data['first_name']} {record.extra_data['last_name']}",
"phone_number": record.extra_data["phone_number"],
"location": {
"country": "NG", # ensure country comes in full version e.g Nigeria
"address": record.extra_data["contact_address"],
},
}
)
mock_quickbooks.create_sales_receipt.assert_called_once_with(
{"id": "23", "name": "Danny Novaka"},
{
"currency": record.currency,
"description": record.description,
"price": record.price,
"amount": record.price,
"discount": 0,
},
)
mock_post.assert_called_once_with(
settings.AUTH_ENDPOINT + "/save-custom-data",
json={
"user_id": record.user,
"data": {
"quickbooks_customer_details": {"id": "23", "name": "Danny Novaka"}
},
},
)
mock_mail.assert_called_once_with(
"first_resume_download", {"first_name": "Danny"}, [record.extra_data["email"]]
)
Я бы сказал, что это был приятный опыт, используя Pteest и использовал бы его больше в других проектах, на которых я работаю.
Количество изменений было минимально для повышения производительности, которые будут получены.
Надеюсь, этот пост подчеркивает, как вы можете постепенно сделать своими меньшими представлениями асинхронных с Starlette Отказ