Merge pull request #7 from ubclaunchpad/process_file_endpoint

POST/PATCH file to API endpoint

.gitignore

@@ -1 +1,2 @@
-env/
+env/
+**/__pycache__

core/i18nilize/migrations/0002_translation.py

@@ -0,0 +1,24 @@
+# Generated by Django 5.1.1 on 2024-10-18 02:44
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('i18nilize', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Translation',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('original_word', models.CharField(max_length=255)),
+                ('translated_word', models.CharField(max_length=255)),
+                ('language', models.CharField(max_length=255)),
+                ('token', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='i18nilize.token')),
+            ],
+        ),
+    ]

core/i18nilize/services/translation_processor.py

@@ -0,0 +1,158 @@
+from django.db import transaction
+from ..models import Translation
+
+"""
+Utility functions for translation file processing.
+"""
+
+
+def validate_translations_data(translations_data):
+    """
+    Light validation of translation file structure and format
+    """
+    if "translations" not in translations_data:
+        return False
+
+    for translations in translations_data["translations"]:
+        if "language" not in translations:
+            return False
+
+        for key, value in translations.items():
+            if not isinstance(key, str) or not isinstance(value, str):
+                return False
+
+    return True
+
+
+def get_new_translations(translations_data, token):
+    """
+    Returns a set of translations to add to the database. If any translation already
+    exists and is being updated, returns False (use PATCH endpoint to make updates instead).
+    """
+    translations_set, languages_set = extract_translations(translations_data)
+    existing_translations = fetch_existing_translations(token, translations_set, languages_set)
+    return get_post_translations(translations_set, existing_translations)
+
+
+def get_updated_translations(translations_data, token):
+    """
+    Returns a set of translations to update in the database. If there are any new translations
+    in the translations list, returns False (use POST endpoint to make new translations instead).
+    """
+    translations_set, languages_set = extract_translations(translations_data)
+    existing_translations = fetch_existing_translations(token, translations_set, languages_set)
+    return get_patch_translations(translations_set, existing_translations)
+
+
+def extract_translations(translations_data):
+    """
+    Extracts translations from the request and returns sets of translations and languages.
+    """
+    translations_set = set()
+    languages_set = set()
+    for translations in translations_data["translations"]:
+        language = translations["language"]
+        languages_set.add(language)
+
+        for original_word, translated_word in translations.items():
+            if original_word == "language":
+                continue
+            translations_set.add((original_word, translated_word, language))
+
+    return translations_set, languages_set
+
+
+def fetch_existing_translations(token, translations_set, languages_set):
+    """
+    Fetches existing translations from database in bulk to reduce number of queries.
+    """
+    existing_translations = {
+        (t.original_word, t.language): t.translated_word
+
+        for t in Translation.objects.filter(
+            token=token,
+            language__in=list(languages_set),
+            original_word__in=[original_word for original_word, _, _ in translations_set],
+        )
+    }
+    return existing_translations
+
+
+def get_post_translations(translations_set, existing_translations):
+    """
+    Compares translations received in request with translations in database.
+    If a translation already exists and is being updated, return False.
+    Otherwise, returns a list of new translations to add to database.
+    """
+    new_translations = []
+    for original_word, translated_word, language in translations_set:
+        key = (original_word, language)
+        if key in existing_translations:
+            if existing_translations[key] == translated_word:
+                continue
+            # Translation already exists and is being updated
+            return False
+        else:
+            new_translations.append((original_word, translated_word, language))
+    return new_translations
+
+
+def get_patch_translations(translations_set, existing_translations):
+    """
+    Compares translations received in request with translations in database.
+    If a translation already exists and is being updated, return False.
+    Otherwise, returns a list of new translations to add to database.
+    """
+    new_translations = []
+
+    for original_word, translated_word, language in translations_set:
+        key = (original_word, language)
+        if key in existing_translations:
+            if existing_translations[key] != translated_word:
+                new_translations.append((original_word, translated_word, language))
+        else:
+            return False
+
+    return new_translations
+
+
+def bulk_create_translations(token, new_translations):
+    """
+    Adds new translations to the database with an atomic transaction.
+    If any addition fails, it will rollback all previous additions i.e database
+    will be unchanged.
+    """
+    bulk_translations = [
+        Translation(
+            token=token,
+            original_word=original_word,
+            translated_word=translated_word,
+            language=language,
+        )
+        for original_word, translated_word, language in new_translations
+    ]
+    try:
+        with transaction.atomic():
+            Translation.objects.bulk_create(bulk_translations)
+        return True, len(bulk_translations)
+    except Exception as e:
+        print(e)
+        return False, 0
+
+
+def bulk_update_translations(token, updated_translations):
+    """
+    Update translations in the database with an atomic transaction.
+    Rollback previous updates if any fail.
+    """
+    try:
+        with transaction.atomic():
+            for original_word, translated_word, language in updated_translations:
+                row = Translation.objects.get(token=token, original_word=original_word, language=language)
+                row.translated_word = translated_word
+                row.save()
+
+        return True, len(updated_translations)
+    except Exception as e:
+        print(e)
+        return False, 0