[#293] add TokenAuth migration

src/openklant/components/token/migrations/0002_tokenauth_identifier_alter_tokenauth_token.py

@@ -0,0 +1,120 @@
+# Generated by Django 4.2.15 on 2024-11-28 13:01
+
+import logging
+
+from typing import Set
+from django.db import migrations, models
+from django.db.migrations.state import StateApps
+from django.db.models.aggregates import Count
+
+from openklant.components.token.utils import generate_token
+
+
+logger = logging.getLogger(__name__)
+
+
+def _generate_unique_identifier(
+    token: "TokenAuth", existing_identifiers: Set[str]
+) -> str:
+    identifier = f"token-{token.pk}"
+    count = token.pk
+
+    while identifier in existing_identifiers:
+        identifier = f"token-{count}"
+
+    return identifier
+
+
+def _generate_unique_identifiers(apps: StateApps, schema_editor) -> None:
+    TokenAuth = apps.get_model("token", "TokenAuth")
+
+    existing_identifiers = set(
+        TokenAuth.objects.values_list("identifier", flat=True)
+    )
+
+    for token in TokenAuth.objects.all():
+        if token.identifier:
+            continue
+
+        identifier = _generate_unique_identifier(token, existing_identifiers)
+
+        logger.debug(f"Generated {identifier} for token {token.pk}")
+
+        token.identifier = identifier
+        token.save(update_fields=("identifier",))
+
+        existing_identifiers.add(identifier)
+
+
+def _generate_unique_token(token: "TokenAuth", existing_tokens: Set[str]) -> str:
+    _token = generate_token()
+
+    while _token in existing_tokens:
+        _token = generate_token()
+
+    return _token
+
+
+def _generate_unique_tokens(apps: StateApps, schema_editor) -> None:
+    TokenAuth = apps.get_model("token", "TokenAuth")
+
+    duplicate_token_values = (
+        TokenAuth.objects.values("token")
+        .annotate(Count("id"))
+        .order_by()
+        .filter(id__count__gt=1)
+    )
+
+    duplicate_tokens = TokenAuth.objects.filter(
+        token__in=[item["token"] for item in duplicate_token_values]
+    )
+
+    existing_tokens = set(
+        TokenAuth.objects.exclude(id__in=[token.id for token in duplicate_tokens])
+        .values_list("token", flat=True)
+    )
+
+    for token in duplicate_tokens:
+        _token = _generate_unique_token(token, existing_tokens)
+
+        logger.debug(f"Generated a new token for {token.pk}")
+
+        token.token = _token
+        token.save(update_fields=("token",))
+
+        existing_tokens.add(token)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("token", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="tokenauth",
+            name="identifier",
+            field=models.CharField(blank=True),
+        ),
+
+        migrations.RunPython(
+            _generate_unique_identifiers,
+            reverse_code=migrations.RunPython.noop
+        ),
+        migrations.RunPython(
+            _generate_unique_tokens,
+            reverse_code=migrations.RunPython.noop
+        ),
+
+        migrations.AlterField(
+            model_name="tokenauth",
+            name="token",
+            field=models.CharField(max_length=40, unique=True, verbose_name="token"),
+        ),
+
+        migrations.AlterField(
+            model_name="tokenauth",
+            name="identifier",
+            field=models.SlugField(unique=True),
+        ),
+    ]

src/openklant/components/token/models.py

@@ -1,12 +1,14 @@
-import binascii
-import os
-
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 
+from openklant.components.token.utils import generate_token
+
 
 class TokenAuth(models.Model):
-    token = models.CharField(_("token"), max_length=40)
+    identifier = models.SlugField(unique=True)
+
+    token = models.CharField(_("token"), max_length=40, unique=True)
+
     contact_person = models.CharField(
         _("contact person"),
         max_length=200,
@@ -51,8 +53,5 @@ def __str__(self):
 
     def save(self, *args, **kwargs):
         if not self.token:
-            self.token = self.generate_token()
+            self.token = generate_token()
         return super().save(*args, **kwargs)
-
-    def generate_token(self):
-        return binascii.hexlify(os.urandom(20)).decode()