Django meets Async

conversions.py

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)

conversions_2.py

async def process_payment(request, order):
  services.process_paystack_payment(request, order)

Dockerfile

CMD gunicorn - workers=4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:5000 run:app - access-logfile -

new_test.py

@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"]]
    )

offending.py

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)

old_test.py

@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"]],
        )

run.py

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}]
)

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

signals.py

@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()

wsgi_to_asgi.py

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