En el desarrollo de aplicaciones web, la gestión de cargas de archivos requiere una validación estricta y, a menudo, la clonación del flujo de datos en memoria para permitir múltiples lecturas sin consumir el stream original. A continuación, se presenta una implementación optimizada en Python utilizando el ecosistema de Django.
Implementación del Controlador de Cargas
El siguiente código encapsula la lógica de validación de contexto, verificación de extensiones permitidas y la duplicación del objeto InMemoryUploadedFile.
import os
from io import BytesIO
from django.core.files.uploadedfile import InMemoryUploadedFile
def validate_and_duplicate_upload(http_request, processing_mode, template_schema=None, type_mapping=None):
"""
Valida el archivo cargado y genera un clon en memoria para su procesamiento.
Args:
http_request: Objeto de solicitud HTTP de Django.
processing_mode: Identificador del tipo de procesamiento.
template_schema: Diccionario con la estructura de la plantilla (opcional).
type_mapping: Mapeo de tipos de campos (opcional).
Returns:
InMemoryUploadedFile: Instancia clonada del archivo o None si falla.
"""
# Validación de contexto para modos específicos
if processing_mode == 'UPLOAD_ASSET' and not (template_schema and type_mapping):
return None
# Extracción y validación básica del archivo
target_file = http_request.FILES.get('document')
if not target_file or not getattr(target_file, 'name', None):
return None
# Verificación de extensión y tipo de objeto
if not _has_valid_extension(target_file.name):
return None
if not isinstance(target_file, InMemoryUploadedFile):
return None
# Clonación del stream de bytes
original_content = target_file.read()
target_file.seek(0) # Restaurar el puntero original
buffer_clone = BytesIO(original_content)
# Construcción del nuevo objeto de archivo
cloned_file = InMemoryUploadedFile(
file=buffer_clone,
field_name=target_file.field_name,
name=target_file.name,
content_type=target_file.content_type,
size=target_file.size,
charset=target_file.charset,
content_type_extra=target_file.content_type_extra
)
return cloned_file
def _has_valid_extension(filename):
"""Verifica si la extensión del archivo está en la lista blanca."""
whitelist = {'.txt', '.pdf', '.doc', '.docx', '.csv'}
_, ext = os.path.splitext(filename)
return ext.lower() in whitelist
Mejoras Arquitectónicas Aplicadas
- Desacoplamiento: Se elimina la dependencia de una clase contenedora, transformando la lógica en una función pura que facilita las pruebas unitarias.
- Cláusulas de Guardia: Se emplean retornos tempranos (early returns) para aplanar la estructura condicional y mejorar la legibilidad.
- Gestión de Streams: Se asegura la restauración del puntero del archivo original mediante
seek(0)después de leer su contenido, evitando efectos secundarios en el objeto de solicitud. - Nomenclatura Descriptiva: Los nombres de variables y funciones reflejan claramente su propósito (
validate_and_duplicate_upload,buffer_clone).
Integración en una Vista
Para utilizar este controlador dentro de una vista de Django, simplemente se invoca la función y se evalúa su resultado:
from django.http import HttpResponse
def my_upload_view(request):
cloned_asset = validate_and_duplicate_upload(
http_request=request,
processing_mode='UPLOAD_ASSET',
template_schema=request.POST.get('schema'),
type_mapping=request.POST.get('mapping')
)
if cloned_asset is None:
return HttpResponse("Error: Carga de archivo inválida o rechazada.", status=400)
# Lógica de negocio con el archivo clonado
asset_name = cloned_asset.name
# ... procesamiento adicional ...