Remove unused functions from billing views

Author: rohitvinnakota-codecov

billing/views.py

@@ -523,232 +523,3 @@ def setup_intent_succeeded(self, setup_intent: stripe.SetupIntent) -> None:
         )
 
     # --- Additional Functions for Robust Flows ---
-
-    def payment_intent_payment_failed(self, payment_intent: stripe.PaymentIntent) -> None:
-        log.error(
-            "Payment Intent Failed",
-            extra=dict(
-                stripe_customer_id=payment_intent.customer,
-                payment_intent_id=payment_intent.id,
-            ),
-        )
-        owners: QuerySet[Owner] = Owner.objects.filter(
-            stripe_customer_id=payment_intent.customer
-        )
-        owners.update(delinquent=True)
-        self._log_updated(list(owners))
-        admins = get_all_admins_for_owners(owners)
-        task_service = TaskService()
-        template_vars = {
-            "amount": getattr(payment_intent, "amount_received", 0) / 100,
-            "date": datetime.now().strftime("%B %-d, %Y"),
-        }
-        for admin in admins:
-            if admin.email:
-                task_service.send_email(
-                    to_addr=admin.email,
-                    subject="Payment Intent Failed",
-                    template_name="payment-intent-failed",
-                    **template_vars,
-                )
-
-    def charge_refunded(self, charge: stripe.Charge) -> None:
-        log.info(
-            "Charge Refunded",
-            extra=dict(
-                stripe_charge_id=charge.id,
-                amount_refunded=charge.amount_refunded,
-            ),
-        )
-        owners: QuerySet[Owner] = Owner.objects.filter(
-            stripe_customer_id=charge.customer
-        )
-        admins = get_all_admins_for_owners(owners)
-        task_service = TaskService()
-        template_vars = {
-            "charge_id": charge.id,
-            "amount_refunded": charge.amount_refunded / 100,
-            "date": datetime.now().strftime("%B %-d, %Y"),
-        }
-        for admin in admins:
-            if admin.email:
-                task_service.send_email(
-                    to_addr=admin.email,
-                    subject="Charge Refunded",
-                    template_name="charge-refunded",
-                    **template_vars,
-                )
-
-    def dispute_created(self, dispute: stripe.Dispute) -> None:
-        log.warning(
-            "Dispute Created",
-            extra=dict(
-                dispute_id=dispute.id,
-                amount=dispute.amount / 100,
-                status=dispute.status,
-            ),
-        )
-        try:
-            charge = stripe.Charge.retrieve(dispute.charge)
-        except Exception as e:
-            log.error(
-                "Failed to retrieve charge for dispute",
-                extra=dict(dispute_id=dispute.id, error=str(e)),
-            )
-            return
-        owners: QuerySet[Owner] = Owner.objects.filter(
-            stripe_customer_id=charge.customer
-        )
-        admins = get_all_admins_for_owners(owners)
-        task_service = TaskService()
-        template_vars = {
-            "dispute_id": dispute.id,
-            "charge_id": dispute.charge,
-            "amount": dispute.amount / 100,
-            "reason": dispute.reason,
-            "date": datetime.now().strftime("%B %-d, %Y"),
-        }
-        for admin in admins:
-            if admin.email:
-                task_service.send_email(
-                    to_addr=admin.email,
-                    subject="Dispute Created on a Charge",
-                    template_name="dispute-created",
-                    **template_vars,
-                )
-
-    def invoice_updated(self, invoice: stripe.Invoice) -> None:
-        log.info(
-            "Invoice Updated",
-            extra=dict(stripe_invoice_id=invoice.id, status=invoice.status),
-        )
-        if invoice.status == "open":
-            owners = Owner.objects.filter(stripe_customer_id=invoice.customer)
-            admins = get_all_admins_for_owners(owners)
-            task_service = TaskService()
-            template_vars = {
-                "amount": invoice.total / 100,
-                "date": datetime.now().strftime("%B %-d, %Y"),
-            }
-            for admin in admins:
-                if admin.email:
-                    task_service.send_email(
-                        to_addr=admin.email,
-                        subject="Invoice Updated",
-                        template_name="invoice-updated",
-                        **template_vars,
-                    )
-
-    def account_updated(self, account: stripe.Account) -> None:
-        log.info(
-            "Account Updated",
-            extra=dict(stripe_account_id=account.id, email=getattr(account, "email", None)),
-        )
-
-    def subscription_payment_retry(self, subscription: stripe.Subscription) -> None:
-        log.info(
-            "Subscription Payment Retry",
-            extra=dict(
-                stripe_subscription_id=subscription.id,
-                stripe_customer_id=subscription.customer,
-            ),
-        )
-        owners: QuerySet[Owner] = Owner.objects.filter(
-            stripe_subscription_id=subscription.id,
-            stripe_customer_id=subscription.customer,
-        )
-        owners.update(delinquent=True)
-        self._log_updated(list(owners))
-        admins = get_all_admins_for_owners(owners)
-        task_service = TaskService()
-        template_vars = {
-            "date": datetime.now().strftime("%B %-d, %Y"),
-            "subscription_id": subscription.id,
-        }
-        for admin in admins:
-            if admin.email:
-                task_service.send_email(
-                    to_addr=admin.email,
-                    subject="Subscription Payment Retry",
-                    template_name="subscription-payment-retry",
-                    **template_vars,
-                )
-
-    def default_event_handler(self, event_object: Any) -> None:
-        log.info(
-            "Default Event Handler: Unhandled event type",
-            extra=dict(event_object=event_object),
-        )
-
-    def simulate_event(self, event_type: str, event_data: dict) -> None:
-        fake_event = type("FakeEvent", (object,), {})()
-        fake_event.type = event_type
-        fake_event.data = type("FakeEventData", (object,), {"object": event_data})()
-        handler = getattr(self, event_type.replace(".", "_"), self.default_event_handler)
-        handler(fake_event.data.object)
-
-    def update_owner_metadata(self, owner: Owner, metadata: dict) -> None:
-        log.info(
-            "Updating owner metadata",
-            extra=dict(ownerid=owner.ownerid, metadata=metadata),
-        )
-        for key, value in metadata.items():
-            setattr(owner, key, value)
-        owner.save()
-
-    def revalidate_subscription(self, subscription: stripe.Subscription) -> None:
-        log.info(
-            "Revalidating subscription",
-            extra=dict(subscription_id=subscription.id),
-        )
-        owners: QuerySet[Owner] = Owner.objects.filter(
-            stripe_subscription_id=subscription.id,
-            stripe_customer_id=subscription.customer,
-        )
-        if not owners.exists():
-            log.error(
-                "Revalidate subscription: No owner found",
-                extra=dict(subscription_id=subscription.id),
-            )
-            return
-        for owner in owners:
-            plan_service = PlanService(current_org=owner)
-            plan_service.update_plan(name="Revalidated Plan", user_count=subscription.quantity)
-        self._log_updated(list(owners))
-
-    def simulate_delay(self, seconds: int) -> None:
-        log.info("Simulating delay", extra={"seconds": seconds})
-        import time
-        time.sleep(seconds)
-
-    def post(self, request: HttpRequest, *args: Any, **kwargs: Any) -> Response:
-        if settings.STRIPE_ENDPOINT_SECRET is None:
-            log.critical(
-                "Stripe endpoint secret improperly configured -- webhooks will not be processed."
-            )
-        try:
-            self.event = stripe.Webhook.construct_event(
-                self.request.body,
-                self.request.META.get(StripeHTTPHeaders.SIGNATURE),
-                settings.STRIPE_ENDPOINT_SECRET,
-            )
-        except stripe.SignatureVerificationError as e:
-            log.warning(f"Stripe webhook event received with invalid signature -- {e}")
-            return Response("Invalid signature", status=status.HTTP_400_BAD_REQUEST)
-        if self.event.type not in StripeWebhookEvents.subscribed_events:
-            log.warning(
-                "Unsupported Stripe webhook event received, exiting",
-                extra=dict(stripe_webhook_event=self.event.type),
-            )
-            return Response("Unsupported event type", status=204)
-
-        log.info(
-            "Stripe webhook event received",
-            extra=dict(stripe_webhook_event=self.event.type),
-        )
-        handler = getattr(self, self.event.type.replace(".", "_"), self.default_event_handler)
-        try:
-            handler(self.event.data.object)
-        except Exception as e:
-            log.error("Error handling event", extra={"error": str(e)})
-        return Response(status=status.HTTP_204_NO_CONTENT)