Initial

django_translatable_fields/operations.py

@@ -0,0 +1,129 @@
+# -*- coding: utf-8 -*-
+"""
+Django Translatable Fields - Migration Operations
+
+This module provides custom Django migration operations for handling
+translatable field data transitions when the translatable parameter
+changes from True to False or vice versa.
+
+Key Features:
+- Automatic data conversion between JSON and string formats
+- Support for translate=True to translate=False transitions
+- Support for translate=False to translate=True transitions
+- Language-aware data extraction and wrapping
+- Reversible migration operations
+- Database-safe SQL operations
+
+Author: Holger Sielaff <holger@backender.de>
+Version: 0.1.0
+"""
+
+import json
+from django.db import migrations
+from django.utils.translation import get_language
+
+
+class ConvertTranslatableField(migrations.Operation):
+    """
+    Custom migration operation to convert between translatable and non-translatable formats
+    """
+    
+    def __init__(self, model_name, field_name, from_translatable, to_translatable, language=None):
+        self.model_name = model_name
+        self.field_name = field_name  
+        self.from_translatable = from_translatable
+        self.to_translatable = to_translatable
+        self.language = language or get_language() or 'en'
+        
+    def state_forwards(self, app_label, state):
+        # No state changes needed - field definition stays the same
+        pass
+        
+    def database_forwards(self, app_label, schema_editor, from_state, to_state):
+        """Convert data when applying migration"""
+        # Get the model
+        from_model = from_state.apps.get_model(app_label, self.model_name)
+        
+        if self.from_translatable and not self.to_translatable:
+            # Convert from JSON to string (extract current language)
+            self._convert_json_to_string(from_model, schema_editor)
+        elif not self.from_translatable and self.to_translatable:
+            # Convert from string to JSON (wrap in language dict)
+            self._convert_string_to_json(from_model, schema_editor)
+            
+    def database_backwards(self, app_label, schema_editor, from_state, to_state):
+        """Convert data when reversing migration"""
+        # Reverse the conversion
+        to_model = to_state.apps.get_model(app_label, self.model_name)
+        
+        if self.from_translatable and not self.to_translatable:
+            # Reverse: string back to JSON
+            self._convert_string_to_json(to_model, schema_editor)
+        elif not self.from_translatable and self.to_translatable:
+            # Reverse: JSON back to string
+            self._convert_json_to_string(to_model, schema_editor)
+    
+    def _convert_json_to_string(self, model, schema_editor):
+        """Convert JSON translations to single language string"""
+        print(f"Converting {self.model_name}.{self.field_name} from JSON to string (language: {self.language})")
+        
+        for obj in model.objects.all():
+            field_value = getattr(obj, self.field_name)
+            
+            if field_value:
+                try:
+                    if isinstance(field_value, str):
+                        # Try to parse JSON
+                        translations = json.loads(field_value)
+                        if isinstance(translations, dict):
+                            # Extract value for current language or fallback
+                            new_value = (
+                                translations.get(self.language) or 
+                                translations.get('en') or 
+                                next(iter(translations.values()), '')
+                            )
+                            
+                            # Update using raw SQL to avoid field processing
+                            table_name = obj._meta.db_table
+                            schema_editor.execute(
+                                f"UPDATE {table_name} SET {self.field_name} = %s WHERE id = %s",
+                                [new_value, obj.pk]
+                            )
+                            print(f"  Converted {obj.pk}: {field_value} -> {new_value}")
+                        
+                except (json.JSONDecodeError, TypeError):
+                    # Already a string, leave as-is
+                    pass
+                    
+    def _convert_string_to_json(self, model, schema_editor):
+        """Convert single language string to JSON translations"""
+        print(f"Converting {self.model_name}.{self.field_name} from string to JSON (language: {self.language})")
+        
+        for obj in model.objects.all():
+            field_value = getattr(obj, self.field_name)
+            
+            if field_value:
+                try:
+                    # Check if it's already JSON
+                    json.loads(field_value)
+                    # If parsing succeeds, it's already JSON - skip
+                except (json.JSONDecodeError, TypeError):
+                    # It's a plain string, convert to JSON
+                    new_value = json.dumps({self.language: str(field_value)})
+                    
+                    # Update using raw SQL to avoid field processing
+                    table_name = obj._meta.db_table  
+                    schema_editor.execute(
+                        f"UPDATE {table_name} SET {self.field_name} = %s WHERE id = %s",
+                        [new_value, obj.pk]
+                    )
+                    print(f"  Converted {obj.pk}: {field_value} -> {new_value}")
+    
+    def describe(self):
+        direction = "translatable" if self.to_translatable else "non-translatable"
+        return f"Convert {self.model_name}.{self.field_name} to {direction}"
+        
+    @property
+    def migration_name_fragment(self):
+        direction = "translatable" if self.to_translatable else "nontranslatable"  
+        return f"convert_{self.field_name}_to__{direction}"