Add API for redeeming enrollment codes for org membership (#2866)

Author: jkachel

b2b/admin.py

@@ -2,7 +2,32 @@
 
 from django.contrib import admin
 
-from b2b.models import ContractPage, OrganizationPage
+from b2b.models import (
+    ContractPage,
+    DiscountContractAttachmentRedemption,
+    OrganizationPage,
+)
+
+
+@admin.register(DiscountContractAttachmentRedemption)
+class DiscountContractAttachmentRedemptionAdmin(admin.ModelAdmin):
+    """Admin for discount attachments."""
+
+    list_display = ["user", "contract", "discount", "created_on"]
+    date_hierarchy = "created_on"
+    fields = ["user", "contract", "discount", "created_on"]
+    readonly_fields = ["user", "contract", "discount", "created_on"]
+
+    def has_add_permission(self, request):  # noqa: ARG002
+        """Disable create."""
+
+        return False
+
+    def has_delete_permission(self, request, obj=None):  # noqa: ARG002
+        """Disable deletions."""
+
+        return False
+
 
 admin.site.register(OrganizationPage)
 admin.site.register(ContractPage)

b2b/migrations/0005_add_discount_attachment_model.py

@@ -0,0 +1,59 @@
+# Generated by Django 4.2.22 on 2025-08-06 20:13
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("ecommerce", "0036_alter_order_state"),
+        ("b2b", "0004_add_sso_org_id_field"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="DiscountContractAttachmentRedemption",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created_on", models.DateTimeField(auto_now_add=True)),
+                ("updated_on", models.DateTimeField(auto_now=True)),
+                (
+                    "contract",
+                    models.ForeignKey(
+                        help_text="The contract that the user was attached to.",
+                        on_delete=django.db.models.deletion.DO_NOTHING,
+                        to="b2b.contractpage",
+                    ),
+                ),
+                (
+                    "discount",
+                    models.ForeignKey(
+                        help_text="The discount that was redemeed.",
+                        on_delete=django.db.models.deletion.DO_NOTHING,
+                        to="ecommerce.discount",
+                    ),
+                ),
+                (
+                    "user",
+                    models.ForeignKey(
+                        help_text="The user that redeemed the discount.",
+                        on_delete=django.db.models.deletion.DO_NOTHING,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "abstract": False,
+            },
+        ),
+    ]

b2b/models.py

@@ -1,10 +1,12 @@
 """Models for B2B data."""
 
+from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.contenttypes.models import ContentType
 from django.db import models
 from django.http import Http404
 from django.utils.text import slugify
+from mitol.common.models import TimestampedModel
 from mitol.common.utils import now_in_utc
 from wagtail.admin.panels import FieldPanel, MultiFieldPanel
 from wagtail.fields import RichTextField
@@ -221,6 +223,24 @@ def get_learners(self):
             .distinct()
         )
 
+    def is_full(self):
+        """Determine if the contract is full or not."""
+
+        return (
+            self.get_learners().count() >= self.max_learners
+            if self.max_learners
+            else False
+        )
+
+    def is_overfull(self):
+        """Determine if the contract is overcommitted."""
+
+        return (
+            self.get_learners().count() > self.max_learners
+            if self.max_learners
+            else False
+        )
+
     def get_course_runs(self):
         """Get the runs associated with the contract."""
 
@@ -250,3 +270,23 @@ def get_discounts(self):
         from ecommerce.models import Discount
 
         return Discount.objects.filter(products__product__in=self.get_products()).all()
+
+
+class DiscountContractAttachmentRedemption(TimestampedModel):
+    """Records when a discount was used to attach the user to a contract."""
+
+    discount = models.ForeignKey(
+        "ecommerce.Discount",
+        on_delete=models.DO_NOTHING,
+        help_text="The discount that was redemeed.",
+    )
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        on_delete=models.DO_NOTHING,
+        help_text="The user that redeemed the discount.",
+    )
+    contract = models.ForeignKey(
+        ContractPage,
+        on_delete=models.DO_NOTHING,
+        help_text="The contract that the user was attached to.",
+    )

b2b/views/v0/__init__.py

@@ -1,23 +1,29 @@
 """Views for the B2B API (v0)."""
 
 from django.contrib.contenttypes.models import ContentType
+from django.db.models import Q
 from django.views.decorators.csrf import csrf_exempt
 from drf_spectacular.utils import extend_schema
+from mitol.common.utils.datetime import now_in_utc
 from rest_framework import status, viewsets
 from rest_framework.permissions import IsAdminUser, IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.views import APIView
 from rest_framework_api_key.permissions import HasAPIKey
 
 from b2b.api import create_b2b_enrollment
-from b2b.models import ContractPage, OrganizationPage
+from b2b.models import (
+    ContractPage,
+    DiscountContractAttachmentRedemption,
+    OrganizationPage,
+)
 from b2b.serializers.v0 import (
     ContractPageSerializer,
     CreateB2BEnrollmentSerializer,
     OrganizationPageSerializer,
 )
 from courses.models import CourseRun
-from ecommerce.models import Product
+from ecommerce.models import Discount, Product
 from main.constants import USER_MSG_TYPE_B2B_ENROLL_SUCCESS
 
 
@@ -74,3 +80,70 @@ def post(self, request, readable_id: str, format=None):  # noqa: A002, ARG002
             if response["result"] == USER_MSG_TYPE_B2B_ENROLL_SUCCESS
             else status.HTTP_406_NOT_ACCEPTABLE,
         )
+
+
+class AttachContractApi(APIView):
+    """View for attaching a user to a B2B contract."""
+
+    permission_classes = [IsAuthenticated]
+
+    @extend_schema(
+        request=None,
+        responses=ContractPageSerializer(many=True),
+    )
+    @csrf_exempt
+    def post(self, request, enrollment_code: str, format=None):  # noqa: A002, ARG002
+        """
+        Use the provided enrollment code to attach the user to a B2B contract.
+
+        This will not create an order, nor will it enroll the user. It will
+        attach the user to the contract and log that the code was used for this
+        purpose (but will _not_ invalidate the code, since we're not actually
+        using it at this point).
+
+        This will respect the activation and expiration dates (of both the contract
+        and the discount), and will make sure there's sufficient available seats
+        in the contract.
+
+        If the user is already in the contract, then we skip it.
+
+        Returns:
+        - list of ContractPageSerializer - the contracts for the user
+        """
+
+        now = now_in_utc()
+        try:
+            code = (
+                Discount.objects.filter(
+                    Q(activation_date__isnull=True) | Q(activation_date__lte=now)
+                )
+                .filter(Q(expiration_date__isnull=True) | Q(expiration_date__gte=now))
+                .get(discount_code=enrollment_code)
+            )
+        except Discount.DoesNotExist:
+            return Response(
+                ContractPageSerializer(request.user.b2b_contracts.all(), many=True).data
+            )
+
+        contract_ids = list(code.b2b_contracts().values_list("id", flat=True))
+        contracts = (
+            ContractPage.objects.filter(pk__in=contract_ids)
+            .exclude(pk__in=request.user.b2b_contracts.all())
+            .exclude(Q(contract_end__lt=now) | Q(contract_start__gt=now))
+            .all()
+        )
+
+        for contract in contracts:
+            if contract.is_full():
+                continue
+
+            request.user.b2b_contracts.add(contract)
+            DiscountContractAttachmentRedemption.objects.create(
+                user=request.user, discount=code, contract=contract
+            )
+
+        request.user.save()
+
+        return Response(
+            ContractPageSerializer(request.user.b2b_contracts.all(), many=True).data
+        )

b2b/views/v0/urls.py

@@ -3,7 +3,12 @@
 from django.urls import include, path, re_path
 from rest_framework.routers import SimpleRouter
 
-from b2b.views.v0 import ContractPageViewSet, Enroll, OrganizationPageViewSet
+from b2b.views.v0 import (
+    AttachContractApi,
+    ContractPageViewSet,
+    Enroll,
+    OrganizationPageViewSet,
+)
 
 app_name = "b2b"
 
@@ -22,4 +27,9 @@
 urlpatterns = [
     re_path(r"^", include(v0_router.urls)),
     path(r"enroll/<str:readable_id>/", Enroll.as_view()),
+    path(
+        r"attach/<str:enrollment_code>/",
+        AttachContractApi.as_view(),
+        name="attach-user",
+    ),
 ]