Created
October 23, 2018 05:04
-
-
Save gbozee/041a832c436cd60305a9b31ea62851b1 to your computer and use it in GitHub Desktop.
Django meets Async
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
async def process_payment(request, order): | |
services.process_paystack_payment(request, order) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
CMD gunicorn - workers=4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:5000 run:app - access-logfile - |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
@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"]] | |
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
@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"]], | |
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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}] | |
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
@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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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] | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment