from typing import Any, Union from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.models import ContentType from django.db import models from django.db.models import CASCADE, SET_NULL from corporate.models.customers import Customer class Event(models.Model): stripe_event_id = models.CharField(max_length=255) type = models.CharField(max_length=255) RECEIVED = 1 EVENT_HANDLER_STARTED = 30 EVENT_HANDLER_FAILED = 40 EVENT_HANDLER_SUCCEEDED = 50 status = models.SmallIntegerField(default=RECEIVED) content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.PositiveIntegerField(db_index=True) content_object = GenericForeignKey("content_type", "object_id") handler_error = models.JSONField(default=None, null=True) def get_event_handler_details_as_dict(self) -> dict[str, Any]: details_dict = {} details_dict["status"] = { Event.RECEIVED: "not_started", Event.EVENT_HANDLER_STARTED: "started", Event.EVENT_HANDLER_FAILED: "failed", Event.EVENT_HANDLER_SUCCEEDED: "succeeded", }[self.status] if self.handler_error: details_dict["error"] = self.handler_error return details_dict def get_last_associated_event_by_type( content_object: Union["Invoice", "PaymentIntent", "Session"], event_type: str ) -> Event | None: content_type = ContentType.objects.get_for_model(type(content_object)) return Event.objects.filter( content_type=content_type, object_id=content_object.id, type=event_type ).last() class Session(models.Model): customer = models.ForeignKey(Customer, on_delete=CASCADE) stripe_session_id = models.CharField(max_length=255, unique=True) CARD_UPDATE_FROM_BILLING_PAGE = 40 CARD_UPDATE_FROM_UPGRADE_PAGE = 50 type = models.SmallIntegerField() CREATED = 1 COMPLETED = 10 status = models.SmallIntegerField(default=CREATED) # Did the user opt to manually manage licenses before clicking on update button? is_manual_license_management_upgrade_session = models.BooleanField(default=False) # CustomerPlan tier that the user is upgrading to. tier = models.SmallIntegerField(null=True) def get_status_as_string(self) -> str: return {Session.CREATED: "created", Session.COMPLETED: "completed"}[self.status] def get_type_as_string(self) -> str: return { Session.CARD_UPDATE_FROM_BILLING_PAGE: "card_update_from_billing_page", Session.CARD_UPDATE_FROM_UPGRADE_PAGE: "card_update_from_upgrade_page", }[self.type] def to_dict(self) -> dict[str, Any]: session_dict: dict[str, Any] = {} session_dict["status"] = self.get_status_as_string() session_dict["type"] = self.get_type_as_string() session_dict["is_manual_license_management_upgrade_session"] = ( self.is_manual_license_management_upgrade_session ) session_dict["tier"] = self.tier event = self.get_last_associated_event() if event is not None: session_dict["event_handler"] = event.get_event_handler_details_as_dict() return session_dict def get_last_associated_event(self) -> Event | None: if self.status == Session.CREATED: return None return get_last_associated_event_by_type(self, "checkout.session.completed") class PaymentIntent(models.Model): # nocoverage customer = models.ForeignKey(Customer, on_delete=CASCADE) stripe_payment_intent_id = models.CharField(max_length=255, unique=True) REQUIRES_PAYMENT_METHOD = 1 REQUIRES_CONFIRMATION = 20 REQUIRES_ACTION = 30 PROCESSING = 40 REQUIRES_CAPTURE = 50 CANCELLED = 60 SUCCEEDED = 70 status = models.SmallIntegerField() last_payment_error = models.JSONField(default=None, null=True) @classmethod def get_status_integer_from_status_text(cls, status_text: str) -> int: return getattr(cls, status_text.upper()) def get_status_as_string(self) -> str: return { PaymentIntent.REQUIRES_PAYMENT_METHOD: "requires_payment_method", PaymentIntent.REQUIRES_CONFIRMATION: "requires_confirmation", PaymentIntent.REQUIRES_ACTION: "requires_action", PaymentIntent.PROCESSING: "processing", PaymentIntent.REQUIRES_CAPTURE: "requires_capture", PaymentIntent.CANCELLED: "cancelled", PaymentIntent.SUCCEEDED: "succeeded", }[self.status] def get_last_associated_event(self) -> Event | None: if self.status == PaymentIntent.SUCCEEDED: event_type = "payment_intent.succeeded" # TODO: Add test for this case. Not sure how to trigger naturally. else: # nocoverage return None # nocoverage return get_last_associated_event_by_type(self, event_type) def to_dict(self) -> dict[str, Any]: payment_intent_dict: dict[str, Any] = {} payment_intent_dict["status"] = self.get_status_as_string() event = self.get_last_associated_event() if event is not None: payment_intent_dict["event_handler"] = event.get_event_handler_details_as_dict() return payment_intent_dict class Invoice(models.Model): customer = models.ForeignKey(Customer, on_delete=CASCADE) stripe_invoice_id = models.CharField(max_length=255, unique=True) plan = models.ForeignKey("CustomerPlan", null=True, default=None, on_delete=SET_NULL) is_created_for_free_trial_upgrade = models.BooleanField(default=False) SENT = 1 PAID = 2 VOID = 3 status = models.SmallIntegerField() def get_status_as_string(self) -> str: return { Invoice.SENT: "sent", Invoice.PAID: "paid", Invoice.VOID: "void", }[self.status] def get_last_associated_event(self) -> Event | None: if self.status == Invoice.PAID: event_type = "invoice.paid" # TODO: Add test for this case. Not sure how to trigger naturally. else: # nocoverage return None # nocoverage return get_last_associated_event_by_type(self, event_type) def to_dict(self) -> dict[str, Any]: stripe_invoice_dict: dict[str, Any] = {} stripe_invoice_dict["status"] = self.get_status_as_string() event = self.get_last_associated_event() if event is not None: stripe_invoice_dict["event_handler"] = event.get_event_handler_details_as_dict() return stripe_invoice_dict