"""This file and its contents are licensed under the Apache License 2.0. Please see the included NOTICE for copyright information and LICENSE for a copy of the license. """ import os from django.core.exceptions import ValidationError as DjangoValidationError # type: ignore[import] from io_storages.localfiles.models import ( LocalFilesExportStorage, LocalFilesImportStorage, normalize_storage_path, ) from io_storages.serializers import ExportStorageSerializer, ImportStorageSerializer from rest_framework import serializers # type: ignore[import] from rest_framework.exceptions import ValidationError as DRFValidationError # type: ignore[import] def _stringify_detail(detail): """Convert DRF/Django validation detail into plain strings for the UI.""" if isinstance(detail, dict): return {key: _stringify_detail(value) for key, value in detail.items()} if isinstance(detail, (list, tuple)): return [_stringify_detail(item) for item in detail] return str(detail) class LocalFilesImportStorageSerializer(ImportStorageSerializer): type = serializers.ReadOnlyField(default=os.path.basename(os.path.dirname(__file__))) class Meta: model = LocalFilesImportStorage fields = '__all__' def validate(self, data): # Validate local file path data = super(LocalFilesImportStorageSerializer, self).validate(data) if 'path' in data: data['path'] = normalize_storage_path(data['path']) storage = LocalFilesImportStorage(**data) try: storage.validate_connection() except (DjangoValidationError, DRFValidationError) as exc: detail = getattr(exc, 'detail', getattr(exc, 'messages', str(exc))) raise DRFValidationError(_stringify_detail(detail)) except Exception as exc: raise DRFValidationError(str(exc)) return data class LocalFilesExportStorageSerializer(ExportStorageSerializer): type = serializers.ReadOnlyField(default=os.path.basename(os.path.dirname(__file__))) class Meta: model = LocalFilesExportStorage fields = '__all__' def validate(self, data): # Validate local file path data = super(LocalFilesExportStorageSerializer, self).validate(data) if 'path' in data: data['path'] = normalize_storage_path(data['path']) storage = LocalFilesExportStorage(**data) try: storage.validate_connection() except (DjangoValidationError, DRFValidationError) as exc: detail = getattr(exc, 'detail', getattr(exc, 'messages', str(exc))) raise DRFValidationError(_stringify_detail(detail)) except Exception as exc: raise DRFValidationError(str(exc)) return data