Try to enhance the drf connectivity

This commit is contained in:
Holger Sielaff
2026-01-12 12:50:11 +01:00
parent 6947b70764
commit 6e58334190

View File

@@ -20,18 +20,31 @@ Usage:
fields = ['id', 'name', 'description', 'price']
# API Usage:
# GET /api/products/?language=de -> Returns German translations
# GET /api/products/?lang=de -> Returns German translations as strings
# GET /api/products/ -> Returns all translations as dicts
# POST /api/products/ with {"name": "Test", "language": "de"} -> Saves to German
# POST /api/products/ with {"name": "Test", "lang": "de"} -> Saves to German
Author: Holger Sielaff <holger@backender.de>
Version: 0.1.0
"""
import json
import logging
from rest_framework import serializers
from django.utils.translation import get_language
# Optional drf-spectacular support for OpenAPI schema generation
try:
from drf_spectacular.utils import extend_schema_field
HAS_SPECTACULAR = True
except ImportError:
HAS_SPECTACULAR = False
# Dummy decorator if drf-spectacular is not installed
def extend_schema_field(schema):
def decorator(cls):
return cls
return decorator
class TranslatableSerializerMixin:
"""
@@ -42,7 +55,7 @@ class TranslatableSerializerMixin:
writing translatable content via API endpoints.
Features:
- Language parameter detection (?language=de)
- Language parameter detection (?lang=de or ?language=de for backward compatibility)
- Automatic field conversion for translatable fields
- Support for both single-language and multi-language responses
- Proper handling of JSON storage format
@@ -54,9 +67,9 @@ class TranslatableSerializerMixin:
fields = ['name', 'description'] # name and description are translatable
# API calls:
# GET /api/items/?language=de -> returns German values
# GET /api/items/?lang=de -> returns German values as strings
# GET /api/items/ -> returns full translation dicts
# POST /api/items/ with language param -> saves to specific language
# POST /api/items/ with {"name": "Test", "lang": "de"} -> saves to specific language
"""
def __init__(self, *args, **kwargs):
@@ -67,7 +80,12 @@ class TranslatableSerializerMixin:
"""Identify and setup translatable fields in the serializer"""
self._translatable_fields = []
if hasattr(self, 'Meta') and hasattr(self.Meta, 'model'):
# Method 1: Check Meta.translatable_fields (explicit definition)
if hasattr(self, 'Meta') and hasattr(self.Meta, 'translatable_fields'):
self._translatable_fields = list(self.Meta.translatable_fields)
# Method 2: Auto-detect from model fields with translatable=True
elif hasattr(self, 'Meta') and hasattr(self.Meta, 'model'):
model = self.Meta.model
for field_name in self.fields:
if hasattr(model, field_name):
@@ -75,15 +93,21 @@ class TranslatableSerializerMixin:
if hasattr(model_field, 'translatable') and model_field.translatable:
self._translatable_fields.append(field_name)
# Replace translatable fields with TranslatableField for proper schema generation
for field_name in self._translatable_fields:
if field_name in self.fields:
self.fields[field_name] = TranslatableField()
def _get_request_language(self):
"""Get the language parameter from the request context"""
if hasattr(self, 'context') and 'request' in self.context:
request = self.context['request']
# Check for language parameter in query params or data
language = request.query_params.get('language')
if not language and hasattr(request, 'data'):
language = request.data.get('language')
return language
# Check for lang parameter in query params or data
# Also support 'language' for backward compatibility
lang = request.query_params.get('lang') or request.query_params.get('language')
if not lang and hasattr(request, 'data'):
lang = request.data.get('lang') or request.data.get('language')
return lang
return None
def to_representation(self, instance):
@@ -139,39 +163,90 @@ class TranslatableSerializerMixin:
return data
def to_internal_value(self, data):
def update(self, instance, validated_data):
"""
Convert API input data to internal representation.
Override update to merge translatable fields for PATCH requests.
Logic:
- With language parameter: converts single values to {language: value}
- Without language parameter:
* If value is dict -> use dict as-is
* If value is not dict -> convert to {'en': value}
For PATCH: merge new translations with existing ones
For PUT: replace completely (default behavior)
"""
request_language = self._get_request_language()
logger = logging.getLogger(__name__)
# Process translatable fields in input data
request_method = None
if hasattr(self, 'context') and 'request' in self.context:
request_method = self.context['request'].method
logger.info(f"UPDATE METHOD CALLED - Version 2.0 - Method: {request_method}")
# For PATCH: merge translatable fields
if request_method == 'PATCH':
for field_name in self._translatable_fields:
if field_name in data:
field_value = data[field_name]
if field_name in validated_data:
new_value = validated_data[field_name]
if request_language:
# Convert single value to translation dict for specific language
if not isinstance(field_value, dict):
data[field_name] = {request_language: str(field_value)}
# If it's already a dict, use it as-is
else:
# No language specified:
# - If dict: use as-is (full translations)
# - If not dict: assume English
if not isinstance(field_value, dict):
data[field_name] = {'en': str(field_value)}
# If it's already a dict, leave it unchanged
# Get RAW value from database using raw SQL (bypass ALL descriptors)
from django.db import connection
model_class = instance.__class__
table_name = model_class._meta.db_table
pk_field = model_class._meta.pk.name
return super().to_internal_value(data)
with connection.cursor() as cursor:
cursor.execute(
f"SELECT {field_name} FROM {table_name} WHERE {pk_field} = %s",
[instance.pk]
)
row = cursor.fetchone()
existing_value = row[0] if row else None
# Parse existing value
existing_translations = {}
if existing_value:
if isinstance(existing_value, str):
try:
existing_translations = json.loads(existing_value)
if not isinstance(existing_translations, dict):
existing_translations = {}
except (json.JSONDecodeError, TypeError):
existing_translations = {}
elif isinstance(existing_value, dict):
existing_translations = existing_value
# Parse new value
new_translations = {}
if new_value:
if isinstance(new_value, str):
try:
new_translations = json.loads(new_value)
if not isinstance(new_translations, dict):
new_translations = {}
except (json.JSONDecodeError, TypeError):
new_translations = {}
elif isinstance(new_value, dict):
new_translations = new_value
# Merge: existing + new (new overwrites)
if existing_translations or new_translations:
merged = existing_translations.copy()
merged.update(new_translations)
validated_data[field_name] = merged
return super().update(instance, validated_data)
@extend_schema_field({
'type': 'object',
'properties': {
'en': {'type': 'string', 'description': 'English translation'},
'de': {'type': 'string', 'description': 'German translation'},
'fr': {'type': 'string', 'description': 'French translation'},
},
'example': {
'en': 'Example text',
'de': 'Beispieltext',
'fr': 'Texte d\'exemple'
},
'description': 'Translatable field: accepts string (sets default language), dict with language codes, or JSON string'
})
class TranslatableField(serializers.Field):
"""
Custom DRF field for handling translatable content.
@@ -224,15 +299,36 @@ class TranslatableField(serializers.Field):
return translations
def to_internal_value(self, data):
"""Convert API input to internal representation"""
if self.language:
# Convert single value to translation dict
"""
Convert API input to internal representation.
- Handles lang parameter for single-value inputs
- Converts to dict format
- Merging for PATCH is handled by TranslatableSerializerMixin.update()
"""
# Get request context from parent serializer
request_language = None
if hasattr(self, 'parent') and self.parent:
# Get lang parameter from request
if hasattr(self.parent, 'context') and 'request' in self.parent.context:
request = self.parent.context['request']
request_language = request.query_params.get('lang') or request.query_params.get('language')
if not request_language and hasattr(request, 'data'):
request_language = request.data.get('lang') or request.data.get('language')
# Use self.language if set explicitly on the field
language = self.language or request_language
# Convert input to translation dict
if language:
# Language specified: convert single value to translation dict
if isinstance(data, dict):
return data
else:
return {self.language: str(data)}
return {language: str(data)}
else:
# Expect full translation dict
# No language: expect full translation dict or assume English
if isinstance(data, dict):
return data
else: