15758223568

Committed 19 Jun 2025 12:50PM UTC coverage: 99.343% (+0.001%) from 99.342%

Build Type

push

github

Committed by GitHub

Commit Message

Webhook fix 4.0.2 (#164)

Coverage Stats

3022 of 3042 relevant lines covered (99.34%)

0.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.16

/checkout/webhook_handler.py

"""Functions that act on receipt of the webhook."""

import json
import logging
import time
from decimal import Decimal

from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.mail import send_mail
from django.http import HttpResponse
from django.template.loader import render_to_string
from django.utils import translation
from django.utils.translation import gettext_lazy as _
from products.models import Product
from stripe import PaymentIntent

from .models import Order, OrderLineItem


def create_order(
    userprofile, billing_details, shipping_details, cart, delivery, pid
):
    """Creates the order if it isn't in the DB."""
    order = Order.objects.create(
        user_profile=userprofile,
        email=billing_details.email,
        shipping_full_name=shipping_details.name,
        shipping_phone_number=shipping_details.phone,
        shipping_country=shipping_details.address.country,
        shipping_postcode=shipping_details.address.postal_code,
        shipping_town_or_city=shipping_details.address.city,
        shipping_street_address_1=shipping_details.address.line1,
        shipping_street_address_2=shipping_details.address.line2,
        shipping_county=shipping_details.address.state,
        billing_full_name=billing_details.name,
        billing_phone_number=billing_details.phone,
        billing_country=billing_details.address.country,
        billing_postcode=billing_details.address.postal_code,
        billing_town_or_city=billing_details.address.city,
        billing_street_address_1=billing_details.address.line1,
        billing_street_address_2=billing_details.address.line2,
        billing_county=billing_details.address.state,
        original_cart=cart,
        delivery_cost=delivery,
        stripe_pid=pid,
    )
    for item_id, item_data in json.loads(cart).items():
        try:
            product = Product.objects.get(id=item_id)
            order_line_item = OrderLineItem(
                order=order,
                product=product,
                quantity=item_data,
            )
            order_line_item.save()
        except Product.DoesNotExist:
            pass
    order.update_total()
    complete_order = Order.objects.get(
        user_profile=userprofile,
        email=billing_details.email,
        shipping_full_name=shipping_details.name,
        shipping_phone_number=shipping_details.phone,
        shipping_country=shipping_details.address.country,
        shipping_postcode=shipping_details.address.postal_code,
        shipping_town_or_city=shipping_details.address.city,
        shipping_street_address_1=shipping_details.address.line1,
        shipping_street_address_2=shipping_details.address.line2,
        shipping_county=shipping_details.address.state,
        billing_full_name=billing_details.name,
        billing_phone_number=billing_details.phone,
        billing_country=billing_details.address.country,
        billing_postcode=billing_details.address.postal_code,
        billing_town_or_city=billing_details.address.city,
        billing_street_address_1=billing_details.address.line1,
        billing_street_address_2=billing_details.address.line2,
        billing_county=billing_details.address.state,
        original_cart=cart,
        stripe_pid=pid,
    )
    return complete_order


def check_order_in_db(
    billing_details, shipping_details, grand_total, cart, pid
):
    """Runs a loop to check if the order was created in the DB.
    If it was, sends and email, if not, returns False."""
    try:
        order = Order.objects.get(
            email__iexact=billing_details.email,
            shipping_full_name__iexact=shipping_details.name,
            shipping_country__iexact=shipping_details.address.country,
            shipping_town_or_city__iexact=shipping_details.address.city,  # noqa E501
            shipping_street_address_1__iexact=shipping_details.address.line1,  # noqa E501
            grand_total=grand_total,
            original_cart=cart,
            stripe_pid=pid,
        )
    except Order.DoesNotExist:
        order = False
    return order


def save_user_info(userprofile, billing_details, shipping_details):
    """Saves user billing and shipping info to the account."""
    userprofile.shipping_full_name = shipping_details.name
    userprofile.shipping_phone_number = shipping_details.phone
    userprofile.shipping_country = shipping_details.address.country
    userprofile.shipping_postcode = shipping_details.address.postal_code  # noqa E501
    userprofile.shipping_town_or_city = shipping_details.address.city  # noqa E501
    userprofile.shipping_street_address_1 = shipping_details.address.line1  # noqa E501
    userprofile.shipping_street_address_2 = shipping_details.address.line2  # noqa E501
    userprofile.shipping_county = shipping_details.address.state
    userprofile.billing_full_name = billing_details.name
    userprofile.billing_phone_number = billing_details.phone
    userprofile.billing_country = billing_details.address.country
    userprofile.billing_postcode = billing_details.address.postal_code  # noqa E501
    userprofile.billing_town_or_city = billing_details.address.city
    userprofile.billing_street_address_1 = billing_details.address.line1  # noqa E501
    userprofile.billing_street_address_2 = billing_details.address.line2  # noqa E501
    userprofile.billing_county = billing_details.address.state
    userprofile.save()


def send_confirmation_email(order, lang):
    """Send the user a confirmation email on a successful order."""
    translation.activate(lang)

    cust_email = order.email
    subject = render_to_string(
        'checkout/confirmation_email/confirmation_email_subject.txt',
        {'order': order},
    )
    body = render_to_string(
        'checkout/confirmation_email/confirmation_email_body.txt',
        {
            'order': order,
            'contact_email': 'connect@nuragicshamanichealing.com',
        },
    )
    body_html = render_to_string(
        'checkout/confirmation_email/confirmation_email_body.html',
        {
            'order': order,
            'contact_email': 'connect@nuragicshamanichealing.com',
        },
    )

    send_mail(
        subject,
        body,
        _(f'NuragicSH Order <{settings.DEFAULT_ORDER_EMAIL}>'),
        [cust_email],
        html_message=body_html,
    )

    translation.activate('en')
    order_subject = render_to_string(
        'checkout/confirmation_email/order_email_subject.txt', {'order': order}
    )
    order_body = render_to_string(
        'checkout/confirmation_email/order_email_body.txt',
        {'order': order, 'contact_email': cust_email},
    )
    order_body_html = render_to_string(
        'checkout/confirmation_email/order_email_body.html',
        {'order': order, 'contact_email': cust_email},
    )

    send_mail(
        order_subject,
        order_body,
        f'NEW ORDER <{settings.DEFAULT_ORDER_EMAIL}>',
        [settings.DEFAULT_ORDER_EMAIL],
        html_message=order_body_html,
    )


def invoice_email(intent, grand_total):
    """Sends an email to confirm an invoice is paid."""
    invoice_id = intent.invoice
    order_subject = render_to_string(
        'checkout/confirmation_email/invoice_email_subject.txt',
        {'invoice_id': invoice_id},
    )
    order_body = render_to_string(
        'checkout/confirmation_email/invoice_email_body.txt',
        {'invoice_id': invoice_id, 'amount': grand_total},
    )

    send_mail(
        order_subject,
        order_body,
        f'INVOICE PAID <{settings.DEFAULT_ORDER_EMAIL}>',
        [settings.DEFAULT_ORDER_EMAIL],
    )


def handle_event(event):
    """Handle a generic/unknown/unexpected webhook event."""
    return HttpResponse(
        content=f'Unhandled Webhook received: {event["type"]}', status=200
    )


def handle_payment_intent_payment_failed(event):
    """Handle the payment_intent.payment_failed webhook from Stripe."""
    return HttpResponse(
        content=f'Webhook received: {event["type"]}', status=200
    )


def handle_payment_intent_succeeded(event):
    """Handle the payment_intent.succeeded webhook from Stripe.
    If the payment is processed, the handler will check for the order,
    if it hasn't been made it will create the order.
    Also saves user info to the profile if selected."""

    # Declares variables for use in the view
    intent = event.data.object
    intent_with_charge = PaymentIntent.retrieve(
        event.data.object.id, expand=['latest_charge']
    )
    # Calculates the correct value for Stripe
    grand_total = round(intent_with_charge.latest_charge.amount / 100, 2)

    if (intent.description) and ('Payment for Invoice' in intent.description):
        invoice_email(intent, grand_total)
        return HttpResponse(
            content=f'Webhook received: {event["type"]} | \
                    SUCCESS: Email sent for Invoice.',
            status=200,
        )

    pid = intent.id
    cart = intent.metadata.cart
    delivery = Decimal(intent.metadata.delivery)
    lang = intent.metadata.lang
    save_info = intent.metadata.save_info
    user = intent.metadata.user
    userprofile = None
    billing_details = intent_with_charge.latest_charge.billing_details
    shipping_details = intent.shipping

    # If the user is logged in it re-declares the user variable.
    if user != 'AnonymousUser':
        user = get_user_model().objects.get(username=intent.metadata.user)
        userprofile = user.userprofile
        # Saves shipping and billing info if requested by the user.
        if save_info:
            save_user_info(userprofile, billing_details, shipping_details)

    # This checks for the order in the Database
    # and runs a loop to see if it is created in the meantime.
    # If the order is found, it breaks the loop.
    attempt = 1
    while attempt <= 6:
        order = check_order_in_db(
            billing_details, shipping_details, grand_total, cart, pid
        )
        if order is False:
            attempt += 1
            time.sleep(1)
        else:
            break

    if order is not False:
        send_confirmation_email(order, lang)
        return HttpResponse(
            content=f'Webhook received: {event["type"]} | \
                    SUCCESS: Verified order already in database',
            status=200,
        )
    # If no order is found, it creates the order.
    try:
        order = create_order(
            userprofile,
            billing_details,
            shipping_details,
            cart,
            delivery,
            pid,
        )
        # Sends a confirmation email after creation
        send_confirmation_email(order, lang)
        return HttpResponse(
            content=f'Webhook received: {event["type"]} | \
                    SUCCESS: Created order in webhook',
            status=200,
        )
    except Exception as error:
        logging.exception(
            'Webhook received: %(type)s | ERROR: %(error)s',
            {'type': event['type'], 'error': error},
        )
        return HttpResponse(status=500)

1	"""Functions that act on receipt of the webhook."""
2
3	import json	1✔
4	import logging	1✔
5	import time	1✔
6	from decimal import Decimal	1✔
7
8	from django.conf import settings	1✔
9	from django.contrib.auth import get_user_model	1✔
10	from django.core.mail import send_mail	1✔
11	from django.http import HttpResponse	1✔
12	from django.template.loader import render_to_string	1✔
13	from django.utils import translation	1✔
14	from django.utils.translation import gettext_lazy as _	1✔
15	from products.models import Product	1✔
16	from stripe import PaymentIntent	1✔
17
18	from .models import Order, OrderLineItem	1✔
19
20
21	def create_order(	1✔
22	userprofile, billing_details, shipping_details, cart, delivery, pid
23	):
24	"""Creates the order if it isn't in the DB."""
25	order = Order.objects.create(	1✔
26	user_profile=userprofile,
27	email=billing_details.email,
28	shipping_full_name=shipping_details.name,
29	shipping_phone_number=shipping_details.phone,
30	shipping_country=shipping_details.address.country,
31	shipping_postcode=shipping_details.address.postal_code,
32	shipping_town_or_city=shipping_details.address.city,
33	shipping_street_address_1=shipping_details.address.line1,
34	shipping_street_address_2=shipping_details.address.line2,
35	shipping_county=shipping_details.address.state,
36	billing_full_name=billing_details.name,
37	billing_phone_number=billing_details.phone,
38	billing_country=billing_details.address.country,
39	billing_postcode=billing_details.address.postal_code,
40	billing_town_or_city=billing_details.address.city,
41	billing_street_address_1=billing_details.address.line1,
42	billing_street_address_2=billing_details.address.line2,
43	billing_county=billing_details.address.state,
44	original_cart=cart,
45	delivery_cost=delivery,
46	stripe_pid=pid,
47	)
48	for item_id, item_data in json.loads(cart).items():	1✔
49	try:	1✔
50	product = Product.objects.get(id=item_id)	1✔
51	order_line_item = OrderLineItem(	×
52	order=order,
53	product=product,
54	quantity=item_data,
55	)
56	order_line_item.save()	×
57	except Product.DoesNotExist:	1✔
58	pass	1✔
59	order.update_total()	1✔
60	complete_order = Order.objects.get(	1✔
61	user_profile=userprofile,
62	email=billing_details.email,
63	shipping_full_name=shipping_details.name,
64	shipping_phone_number=shipping_details.phone,
65	shipping_country=shipping_details.address.country,
66	shipping_postcode=shipping_details.address.postal_code,
67	shipping_town_or_city=shipping_details.address.city,
68	shipping_street_address_1=shipping_details.address.line1,
69	shipping_street_address_2=shipping_details.address.line2,
70	shipping_county=shipping_details.address.state,
71	billing_full_name=billing_details.name,
72	billing_phone_number=billing_details.phone,
73	billing_country=billing_details.address.country,
74	billing_postcode=billing_details.address.postal_code,
75	billing_town_or_city=billing_details.address.city,
76	billing_street_address_1=billing_details.address.line1,
77	billing_street_address_2=billing_details.address.line2,
78	billing_county=billing_details.address.state,
79	original_cart=cart,
80	stripe_pid=pid,
81	)
82	return complete_order	1✔
83
84
85	def check_order_in_db(	1✔
86	billing_details, shipping_details, grand_total, cart, pid
87	):
88	"""Runs a loop to check if the order was created in the DB.
89	If it was, sends and email, if not, returns False."""
90	try:	1✔
91	order = Order.objects.get(	1✔
92	email__iexact=billing_details.email,
93	shipping_full_name__iexact=shipping_details.name,
94	shipping_country__iexact=shipping_details.address.country,
95	shipping_town_or_city__iexact=shipping_details.address.city, # noqa E501
96	shipping_street_address_1__iexact=shipping_details.address.line1, # noqa E501
97	grand_total=grand_total,
98	original_cart=cart,
99	stripe_pid=pid,
100	)
101	except Order.DoesNotExist:	1✔
102	order = False	1✔
103	return order	1✔
104
105
106	def save_user_info(userprofile, billing_details, shipping_details):	1✔
107	"""Saves user billing and shipping info to the account."""
108	userprofile.shipping_full_name = shipping_details.name	1✔
109	userprofile.shipping_phone_number = shipping_details.phone	1✔
110	userprofile.shipping_country = shipping_details.address.country	1✔
111	userprofile.shipping_postcode = shipping_details.address.postal_code # noqa E501	1✔
112	userprofile.shipping_town_or_city = shipping_details.address.city # noqa E501	1✔
113	userprofile.shipping_street_address_1 = shipping_details.address.line1 # noqa E501	1✔
114	userprofile.shipping_street_address_2 = shipping_details.address.line2 # noqa E501	1✔
115	userprofile.shipping_county = shipping_details.address.state	1✔
116	userprofile.billing_full_name = billing_details.name	1✔
117	userprofile.billing_phone_number = billing_details.phone	1✔
118	userprofile.billing_country = billing_details.address.country	1✔
119	userprofile.billing_postcode = billing_details.address.postal_code # noqa E501	1✔
120	userprofile.billing_town_or_city = billing_details.address.city	1✔
121	userprofile.billing_street_address_1 = billing_details.address.line1 # noqa E501	1✔
122	userprofile.billing_street_address_2 = billing_details.address.line2 # noqa E501	1✔
123	userprofile.billing_county = billing_details.address.state	1✔
124	userprofile.save()	1✔
125
126
127	def send_confirmation_email(order, lang):	1✔
128	"""Send the user a confirmation email on a successful order."""
129	translation.activate(lang)	1✔
130
131	cust_email = order.email	1✔
132	subject = render_to_string(	1✔
133	'checkout/confirmation_email/confirmation_email_subject.txt',
134	{'order': order},
135	)
136	body = render_to_string(	1✔
137	'checkout/confirmation_email/confirmation_email_body.txt',
138	{
139	'order': order,
140	'contact_email': 'connect@nuragicshamanichealing.com',
141	},
142	)
143	body_html = render_to_string(	1✔
144	'checkout/confirmation_email/confirmation_email_body.html',
145	{
146	'order': order,
147	'contact_email': 'connect@nuragicshamanichealing.com',
148	},
149	)
150
151	send_mail(	1✔
152	subject,
153	body,
154	_(f'NuragicSH Order <{settings.DEFAULT_ORDER_EMAIL}>'),
155	[cust_email],
156	html_message=body_html,
157	)
158
159	translation.activate('en')	1✔
160	order_subject = render_to_string(	1✔
161	'checkout/confirmation_email/order_email_subject.txt', {'order': order}
162	)
163	order_body = render_to_string(	1✔
164	'checkout/confirmation_email/order_email_body.txt',
165	{'order': order, 'contact_email': cust_email},
166	)
167	order_body_html = render_to_string(	1✔
168	'checkout/confirmation_email/order_email_body.html',
169	{'order': order, 'contact_email': cust_email},
170	)
171
172	send_mail(	1✔
173	order_subject,
174	order_body,
175	f'NEW ORDER <{settings.DEFAULT_ORDER_EMAIL}>',
176	[settings.DEFAULT_ORDER_EMAIL],
177	html_message=order_body_html,
178	)
179
180
181	def invoice_email(intent, grand_total):	1✔
182	"""Sends an email to confirm an invoice is paid."""
183	invoice_id = intent.invoice	×
184	order_subject = render_to_string(	×
185	'checkout/confirmation_email/invoice_email_subject.txt',
186	{'invoice_id': invoice_id},
187	)
188	order_body = render_to_string(	×
189	'checkout/confirmation_email/invoice_email_body.txt',
190	{'invoice_id': invoice_id, 'amount': grand_total},
191	)
192
193	send_mail(	×
194	order_subject,
195	order_body,
196	f'INVOICE PAID <{settings.DEFAULT_ORDER_EMAIL}>',
197	[settings.DEFAULT_ORDER_EMAIL],
198	)
199
200
201	def handle_event(event):	1✔
202	"""Handle a generic/unknown/unexpected webhook event."""
203	return HttpResponse(	1✔
204	content=f'Unhandled Webhook received: {event["type"]}', status=200
205	)
206
207
208	def handle_payment_intent_payment_failed(event):	1✔
209	"""Handle the payment_intent.payment_failed webhook from Stripe."""
210	return HttpResponse(	1✔
211	content=f'Webhook received: {event["type"]}', status=200
212	)
213
214
215	def handle_payment_intent_succeeded(event):	1✔
216	"""Handle the payment_intent.succeeded webhook from Stripe.
217	If the payment is processed, the handler will check for the order,
218	if it hasn't been made it will create the order.
219	Also saves user info to the profile if selected."""
220
221	# Declares variables for use in the view
222	intent = event.data.object	1✔
223	intent_with_charge = PaymentIntent.retrieve(	1✔
224	event.data.object.id, expand=['latest_charge']
225	)
226	# Calculates the correct value for Stripe
227	grand_total = round(intent_with_charge.latest_charge.amount / 100, 2)	1✔
228
229	if (intent.description) and ('Payment for Invoice' in intent.description):	1✔
230	invoice_email(intent, grand_total)	×
231	return HttpResponse(	×
232	content=f'Webhook received: {event["type"]} \| \
233	SUCCESS: Email sent for Invoice.',
234	status=200,
235	)
236
237	pid = intent.id	1✔
238	cart = intent.metadata.cart	1✔
239	delivery = Decimal(intent.metadata.delivery)	1✔
240	lang = intent.metadata.lang	1✔
241	save_info = intent.metadata.save_info	1✔
242	user = intent.metadata.user	1✔
243	userprofile = None	1✔
244	billing_details = intent_with_charge.latest_charge.billing_details	1✔
245	shipping_details = intent.shipping	1✔
246
247	# If the user is logged in it re-declares the user variable.
248	if user != 'AnonymousUser':	1✔
249	user = get_user_model().objects.get(username=intent.metadata.user)	1✔
250	userprofile = user.userprofile	1✔
251	# Saves shipping and billing info if requested by the user.
252	if save_info:	1✔
253	save_user_info(userprofile, billing_details, shipping_details)	1✔
254
255	# This checks for the order in the Database
256	# and runs a loop to see if it is created in the meantime.
257	# If the order is found, it breaks the loop.
258	attempt = 1	1✔
259	while attempt <= 6:	1✔
260	order = check_order_in_db(	1✔
261	billing_details, shipping_details, grand_total, cart, pid
262	)
263	if order is False:	1✔
264	attempt += 1	1✔
265	time.sleep(1)	1✔
266	else:
267	break	×
268
269	if order is not False:	1✔
270	send_confirmation_email(order, lang)	×
271	return HttpResponse(	×
272	content=f'Webhook received: {event["type"]} \| \
273	SUCCESS: Verified order already in database',
274	status=200,
275	)
276	# If no order is found, it creates the order.
277	try:	1✔
278	order = create_order(	1✔
279	userprofile,
280	billing_details,
281	shipping_details,
282	cart,
283	delivery,
284	pid,
285	)
286	# Sends a confirmation email after creation
287	send_confirmation_email(order, lang)	1✔
288	return HttpResponse(	1✔
289	content=f'Webhook received: {event["type"]} \| \
290	SUCCESS: Created order in webhook',
291	status=200,
292	)
293	except Exception as error:	×
294	logging.exception(	×
295	'Webhook received: %(type)s \| ERROR: %(error)s',
296	{'type': event['type'], 'error': error},
297	)
298	return HttpResponse(status=500)	×

Ri-Dearg / nuragic-sh / 15758223568

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous