Django-Translatable-Fields/django_translatable_fields/operations.py

# -*- 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}"