Merge pull request #178 from AvaCodeSolutions/feat/159/deliver-contents-job-endpoint

feat: #159 API key management

Author: payamnj

django_email_learning/apps.py

@@ -2,6 +2,9 @@
 from django.core import checks
 
 
+PLATFORM_ADMIN_GROUP_NAME = "Platform Admin"
+
+
 def check_site_base_url_config(app_configs, **kwargs):  # type: ignore[no-untyped-def]
     errors = []
     from django.conf import settings

django_email_learning/decorators.py

@@ -1,6 +1,7 @@
 from functools import wraps
 from django.http import JsonResponse
 from django_email_learning.models import OrganizationUser
+from django_email_learning.apps import PLATFORM_ADMIN_GROUP_NAME
 import typing
 
 
@@ -13,7 +14,9 @@ def _wrapped_view(request, *view_args, **view_kwargs) -> JsonResponse:  # type:
 
             if (
                 not request.user.is_superuser
-                and not request.user.groups.filter(name="Platform Admins").exists()
+                and not request.user.groups.filter(
+                    name=PLATFORM_ADMIN_GROUP_NAME
+                ).exists()
             ):
                 return JsonResponse({"error": "Forbidden"}, status=403)
             return view_func(request, *view_args, **view_kwargs)

django_email_learning/migrations/0002_apikey.py

@@ -0,0 +1,50 @@
+# Generated by Django 6.0.1 on 2026-01-28 10:45
+
+import django.core.validators
+import django.db.models.deletion
+import django_email_learning.models
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_email_learning", "0001_initial"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="ApiKey",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "key",
+                    models.CharField(
+                        max_length=256,
+                        unique=True,
+                        validators=[django.core.validators.MinLengthValidator(50)],
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "created_by",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            bases=(django_email_learning.models.EncryptionMixin, models.Model),
+        ),
+    ]

django_email_learning/models.py

@@ -10,9 +10,13 @@
 from django.core.files.storage import default_storage
 from django.urls import reverse
 from django.db import models, transaction
-from django.core.validators import MaxValueValidator, MinValueValidator
+from django.core.validators import (
+    MaxValueValidator,
+    MinValueValidator,
+    MinLengthValidator,
+)
 from django.core.exceptions import ImproperlyConfigured
-from cryptography.fernet import Fernet
+from cryptography.fernet import Fernet, InvalidToken
 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
 from cryptography.hazmat.primitives import hashes
 from django.forms import ValidationError
@@ -103,24 +107,7 @@ def __str__(self) -> str:
         return f"{self.user.username} - {self.organization.name}"
 
 
-class ImapConnection(models.Model):
-    server = models.CharField(max_length=200, validators=[is_domain_or_ip])
-    port = models.IntegerField(db_default=993)
-    email = models.EmailField(max_length=200, unique=True)
-    password = models.CharField(max_length=200)
-    organization = models.ForeignKey(Organization, on_delete=models.CASCADE)
-
-    def __str__(self) -> str:
-        return f"{self.email}|{self.server}:{self.port}"
-
-    def save(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
-        if self.password:
-            self.password = self._encrypt_password(self.password)
-        if self.server:
-            self.server = self.server.lower()
-        self.full_clean()
-        super().save(*args, **kwargs)
-
+class EncryptionMixin:
     def _fernet(self) -> Fernet:
         kdf = PBKDF2HMAC(
             algorithm=hashes.SHA256(), length=32, salt=FIXED_SALT, iterations=100000
@@ -143,6 +130,29 @@ def decrypt_password(self, encrypted_password: str) -> str:
         return f.decrypt(encrypted_password.encode()).decode()
 
 
+class ImapConnection(EncryptionMixin, models.Model):
+    server = models.CharField(max_length=200, validators=[is_domain_or_ip])
+    port = models.IntegerField(db_default=993)
+    email = models.EmailField(max_length=200, unique=True)
+    password = models.CharField(max_length=200)
+    organization = models.ForeignKey(Organization, on_delete=models.CASCADE)
+
+    def __str__(self) -> str:
+        return f"{self.email}|{self.server}:{self.port}"
+
+    def save(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
+        if self.password:
+            try:
+                self.decrypt_password(self.password)
+                # Password is already encrypted
+            except InvalidToken:
+                self.password = self._encrypt_password(self.password)
+        if self.server:
+            self.server = self.server.lower()
+        self.full_clean()
+        super().save(*args, **kwargs)
+
+
 class Course(models.Model):
     title = models.CharField(max_length=200)
     slug = models.SlugField(
@@ -654,3 +664,30 @@ def save(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
             )
         self.full_clean()
         super().save(*args, **kwargs)
+
+
+class ApiKey(EncryptionMixin, models.Model):
+    key = models.CharField(
+        max_length=256, unique=True, validators=[MinLengthValidator(50)]
+    )
+    created_at = models.DateTimeField(auto_now_add=True)
+    created_by = models.ForeignKey(
+        User, on_delete=models.SET_NULL, null=True, blank=True
+    )
+
+    @classmethod
+    def generate_key(cls) -> str:
+        return (
+            base64.urlsafe_b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes)
+            .decode()
+            .rstrip("=")
+        )
+
+    def save(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
+        try:
+            self.decrypt_password(self.key)
+            # Key is already encrypted
+        except InvalidToken:
+            self.key = self._encrypt_password(self.key)
+        self.full_clean()
+        super().save(*args, **kwargs)

django_email_learning/platform/api/serializers.py

@@ -10,6 +10,7 @@
 from typing import Optional, Literal, Any, Callable
 from django.urls import reverse
 from django_email_learning.models import (
+    ApiKey,
     DeliveryStatus,
     Organization,
     ImapConnection,
@@ -26,6 +27,26 @@
 import enum
 
 
+class ApiKeyResponse(BaseModel):
+    id: int
+    key: str
+    created_at: datetime
+    created_by: Optional[str] = None
+
+    @staticmethod
+    def from_django_model(api_key: ApiKey) -> "ApiKeyResponse":
+        return ApiKeyResponse.model_validate(
+            {
+                "id": api_key.id,  # type: ignore[attr-defined]
+                "key": api_key.decrypt_password(api_key.key),
+                "created_at": api_key.created_at,
+                "created_by": api_key.created_by.username
+                if api_key.created_by
+                else None,
+            }
+        )
+
+
 class CreateCourseRequest(BaseModel):
     title: str = Field(min_length=1, examples=["Introduction to Python"])
     slug: str = Field(

django_email_learning/platform/api/urls.py

@@ -1,6 +1,8 @@
 from django.urls import path
 from django.views.defaults import page_not_found
 from django_email_learning.platform.api.views import (
+    ApiKeyView,
+    SingleApiKeyView,
     CourseView,
     EnrollmentView,
     EnrollmentsStatisticsView,
@@ -81,6 +83,12 @@
         SingleOrganizationView.as_view(),
         name="single_organization_view",
     ),
+    path("api_keys/", ApiKeyView.as_view(), name="api_key_view"),
+    path(
+        "api_keys/<int:api_key_id>/",
+        SingleApiKeyView.as_view(),
+        name="single_api_key_view",
+    ),
     path("session", UpdateSessionView.as_view(), name="update_session_view"),
     path("", page_not_found, name="root"),
 ]

django_email_learning/platform/api/views.py

@@ -15,6 +15,7 @@
 from django_email_learning.platform.api import serializers
 from django_email_learning.platform.api.pagniated_api_mixin import PaginatedApiMixin
 from django_email_learning.models import (
+    ApiKey,
     Course,
     CourseContent,
     Enrollment,
@@ -592,6 +593,48 @@ def get(self, request, *args, **kwargs) -> JsonResponse:  # type: ignore[no-unty
         return JsonResponse({"statistics": stats}, status=200)
 
 
+@method_decorator(is_platform_admin(), name="post")
+@method_decorator(is_platform_admin(), name="get")
+class ApiKeyView(View):
+    def post(self, request, *args, **kwargs) -> JsonResponse:  # type: ignore[no-untyped-def]
+        try:
+            key = ApiKey.generate_key()
+            api_key = ApiKey(key=key, created_by=request.user)
+            api_key.save()
+            return JsonResponse(
+                serializers.ApiKeyResponse.from_django_model(api_key).model_dump(),
+                status=201,
+            )
+        except ValidationError as e:
+            return JsonResponse({"error": e.json()}, status=400)
+        except IntegrityError as e:
+            return JsonResponse({"error": str(e)}, status=409)
+
+    def get(self, request, *args, **kwargs) -> JsonResponse:  # type: ignore[no-untyped-def]
+        api_keys = ApiKey.objects.all()  # type: ignore[attr-defined]
+        response_list = []
+        for api_key in api_keys:
+            response_list.append(
+                serializers.ApiKeyResponse.from_django_model(api_key).model_dump()
+            )
+        return JsonResponse({"api_keys": response_list}, status=200)
+
+
+@method_decorator(is_platform_admin(), name="delete")
+class SingleApiKeyView(View):
+    def delete(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
+        try:
+            api_key = ApiKey.objects.get(id=kwargs["api_key_id"])
+            api_key.delete()
+            return JsonResponse({"message": "API Key deleted successfully"}, status=200)
+        except ApiKey.DoesNotExist:
+            return JsonResponse({"error": "API Key not found"}, status=404)
+        except ValidationError as e:
+            return JsonResponse({"error": e.json()}, status=400)
+        except IntegrityError as e:
+            return JsonResponse({"error": str(e)}, status=409)
+
+
 class RootView(View):
     def get(self, request, *args, **kwargs) -> JsonResponse:  # type: ignore[no-untyped-def]
         return JsonResponse({"message": "Email Learning API is running."}, status=200)

django_email_learning/platform/urls.py

@@ -5,15 +5,17 @@
     Courses,
     Organizations,
     Learners,
+    ApiKeys,
 )
 
-app_name = "email_learning"
+app_name = "django_email_learning"
 
 urlpatterns = [
     path("courses/", Courses.as_view(), name="courses_view"),
     path("courses/<int:course_id>/", CourseView.as_view(), name="course_detail_view"),
     path("organizations/", Organizations.as_view(), name="organizations_view"),
     path("learners/", Learners.as_view(), name="learners_view"),
+    path("settings/api_keys/", ApiKeys.as_view(), name="api_keys_view"),
     path(
         "",
         RedirectView.as_view(

django_email_learning/platform/views.py

@@ -108,3 +108,14 @@ def get_context_data(self, **kwargs):  # type: ignore[no-untyped-def]
         context = super().get_context_data(**kwargs)
         context["page_title"] = _("Learners")
         return context
+
+
+@method_decorator(login_required, name="dispatch")
+@method_decorator(is_platform_admin(), name="dispatch")
+class ApiKeys(BasePlatformView):
+    template_name = "platform/settings_api_keys.html"
+
+    def get_context_data(self, **kwargs):  # type: ignore[no-untyped-def]
+        context = super().get_context_data(**kwargs)
+        context["page_title"] = _("API Keys")
+        return context

django_email_learning/signals.py

@@ -2,6 +2,7 @@
 from django.contrib.contenttypes.models import ContentType
 from django.db.models.signals import post_migrate
 from django.dispatch import receiver
+from django_email_learning.apps import PLATFORM_ADMIN_GROUP_NAME
 
 
 @receiver(post_migrate)
@@ -79,10 +80,10 @@ def create_platform_admin_group(sender, **kwargs) -> None:  # type: ignore[no-un
     )
 
     platform_admin_group, created = Group.objects.get_or_create(
-        name="Email Learning Platform Admins"
+        name=PLATFORM_ADMIN_GROUP_NAME
     )
     platform_admin_group.permissions.set(perms)
-    print("Platform Admin group created.")
+    print(f"{PLATFORM_ADMIN_GROUP_NAME} group created.")
 
 
 @receiver(post_migrate)