membersystem/src/accounting/views.py

"""Views for the membership app."""

import stripe
from django.conf import settings
from django.contrib.sites.models import Site
from django.core.mail import send_mail
from django.db import transaction
from django.http import HttpRequest
from django.http import HttpResponse
from django.shortcuts import get_object_or_404
from django.shortcuts import redirect
from django.shortcuts import render
from django.urls import reverse
from django.views.decorators.csrf import csrf_exempt
from django_view_decorator import namespaced_decorator_factory
from djmoney.money import Money

from . import models

order_view = namespaced_decorator_factory(namespace="order", base_path="order")

stripe.api_key = settings.STRIPE_API_KEY


@order_view(
    paths="<int:order_id>/",
    name="detail",
    login_required=True,
)
def order_detail(request: HttpRequest, order_id: int) -> HttpResponse:
    """View to show the details of a member."""
    user = request.user  # People just need to login to pay something, not necessarily be a member
    order = models.Order.objects.get(pk=order_id, member=user)

    context = {
        "order": order,
    }

    return render(
        request=request,
        template_name="accounting/order/detail.html",
        context=context,
    )


@order_view(
    paths="<int:order_id>/pay/",
    name="pay",
    login_required=True,
)
def order_pay(request: HttpRequest, order_id: int) -> HttpResponse:
    """Create a Stripe session and redirects to Stripe Checkout."""
    user = request.user  # People just need to login to pay something, not necessarily be a member
    order = models.Order.objects.get(pk=order_id, member=user)
    current_site = Site.objects.get_current(request)
    base_domain = f"https://{current_site.domain}"
    if settings.DEBUG:
        f"http://{current_site.domain}"

    try:
        line_items = []
        for item in order.items.all():
            line_items.append(  # noqa: PERF401
                {
                    "price_data": {
                        "currency": item.total_with_vat.currency,
                        "unit_amount": int((item.price + item.vat).amount * 100),
                        "product_data": {
                            "name": item.product.name,
                        },
                    },
                    "quantity": item.quantity,
                }
            )
        checkout_session = stripe.checkout.Session.create(
            line_items=line_items,
            metadata={"order_id": order.id},
            mode="payment",
            success_url=base_domain + reverse("order:success", kwargs={"order_id": order.id}),
            cancel_url=base_domain + "/cancel",
        )
    except Exception as e:
        send_mail("Error in checkout", str(e), settings.DEFAULT_FROM_EMAIL, settings.ADMINS)
        raise

    # TODO: Redirect with status=303
    return redirect(checkout_session.url)


@transaction.atomic
@order_view(
    paths="<int:order_id>/pay/success/",
    name="success",
    login_required=True,
)
def success(request: HttpRequest, order_id: int) -> HttpResponse:
    """Create a Stripe session and redirects to Stripe Checkout.

    From Stripe docs: When you have a webhook endpoint set up to listen for checkout.session.completed events and
    you set a success_url, Checkout waits for your server to respond to the webhook event delivery before redirecting
    your customer. If you use this approach, make sure your server responds to checkout.session.completed events as
    quickly as possible.
    """
    user = request.user  # People just need to login to pay something, not necessarily be a member
    order = models.Order.objects.get(pk=order_id, member=user)

    context = {
        "order": order,
    }

    return render(
        request=request,
        template_name="accounting/order/success.html",
        context=context,
    )


@transaction.atomic
@order_view(
    paths="<int:order_id>/pay/cancel/",
    name="cancel",
    login_required=True,
)
def cancel(request: HttpRequest, order_id: int) -> HttpResponse:
    """Page to display when a payment is canceled."""
    user = request.user  # People just need to login to pay something, not necessarily be a member
    order = models.Order.objects.get(pk=order_id, member=user)

    context = {
        "order": order,
    }

    return render(
        request=request,
        template_name="accounting/order/cancel.html",
        context=context,
    )


@transaction.atomic
@order_view(
    paths="stripe/webhook/",
    name="webhook",
)
@csrf_exempt
def stripe_webhook(request: HttpRequest) -> HttpResponse:
    """Handle Stripe webhook.

    https://docs.stripe.com/metadata/use-cases
    """
    payload = request.body
    sig_header = request.headers["stripe-signature"]
    event = None

    try:
        event = stripe.Webhook.construct_event(payload, sig_header, settings.STRIPE_ENDPOINT_SECRET)
    except ValueError:
        # Invalid payload
        return HttpResponse(status=400)
    except stripe.error.SignatureVerificationError:
        # Invalid signature
        return HttpResponse(status=400)

    if event["type"] == "checkout.session.completed" or event["type"] == "checkout.session.async_payment_succeeded":
        # Order is marked paid via signals, Membership is activated via signals.
        order_id = event["data"]["object"]["metadata"]["order_id"]
        order = get_object_or_404(models.Order, pk=order_id)
        if not models.Payment.objects.filter(order=order).exists():
            models.Payment.objects.create(
                order=order,
                amount=Money(event["data"]["object"]["amount_total"], event["data"]["object"]["currency"]),
                description="Paid via Stripe",
                payment_type=models.PaymentType.objects.get_or_create(name="Stripe")[0],
                external_transaction_id=event["id"],
            )

    return HttpResponse(status=200)