file_storage

FileStorageManagerMixin

Bases: ProxyBasePolymorphicManager

Mixin for managers for FileStorage and its proxy models.

Source code in src/apps/files/models/file_storage.py

class FileStorageManagerMixin(ProxyBasePolymorphicManager):
    """Mixin for managers for FileStorage and its proxy models."""

    def get_for_object(self, object: dict) -> models.Model:
        """Get FileStorage for a dict object.

        Validates input object and returns single FileStorage instance
        based on its storage_service value and FileStorage-specific extra values.
        """
        self.model.validate_object(object)
        try:
            return self.get(
                csc_project=object["csc_project"],
                storage_service=object["storage_service"],
            )
        except self.model.DoesNotExist:
            raise exceptions.NotFound()

    def _group_files_by_key(
        self, files: List[dict], raise_exception: bool
    ) -> Dict[tuple, List[dict]]:
        """Group Files by their FileStorage key."""
        files_by_key: Dict[self.model.key_type, List[dict]] = {}
        for f in files:
            try:
                self.model.validate_object(f)
                files_by_key.setdefault(self.model.get_key(f), []).append(f)
            except ValidationError as e:
                if raise_exception:
                    raise e
                f.setdefault("errors", {}).update(e.detail)
                f["storage"] = None
        return files_by_key

    def _get_existing_filestorages_by_key(
        self, files_by_key: Dict[tuple, List[dict]]
    ) -> Dict[tuple, List[dict]]:
        """Get FileStorage instances that exist for files by FileStorage key."""
        storages_by_key: Dict[self.model.key_type, self.model] = {}
        filters = [models.Q(**key._asdict()) for key in files_by_key]
        if filters:
            combined_filters = functools.reduce(operator.or_, filters)
            queryset = self.filter(combined_filters)
            storages_by_key = {self.model.get_key(fs): fs for fs in queryset}
        return storages_by_key

    def _create_missing_filestorages(
        self,
        files_by_key: Dict[tuple, List[dict]],
        storages_by_key: Dict[tuple, List[dict]],
        allow_create,
        raise_exception,
    ) -> Tuple[Dict[tuple, List[dict]], Dict[str, dict]]:
        """Create missing FileStorage instances.

        If files_by_key contains FileStorage keys that don't exist in
        storages_by_key, create them. If creation is not allowed,
        raise exception or return error.

        Returns tuple (modified storages_by_key, errors_by_key)."""
        errors_by_key = {}
        missing_storages = [key for key in files_by_key if key not in storages_by_key]
        for key in missing_storages:
            try:
                if not allow_create:
                    raise ValidationError({"storage_service": "No matching file storage found."})
                key_dict = key._asdict()
                new_storage = self.model.get_proxy_instance(**key_dict)
                new_storage.save()
                storages_by_key[key] = new_storage
            except ValidationError as e:
                if raise_exception:
                    raise e
                errors_by_key[key] = e.detail
        return storages_by_key, errors_by_key

    def assign_to_file_data(
        self,
        files: List[dict],
        allow_create=False,
        raise_exception=True,
        remove_filestorage_fields=False,
    ) -> List[dict]:
        """
        Retrieve FileStorage instances and assign to file data.

        Retrieves (or optionally creates) a FileStorage instance for
        each file in files list. The instance is assigned to
        file['storage'].

        Files are validated for required FileStorage fields based on
        file['storage_service']. When raise_exception is False and
        an error occurs, the error will be added to file['errors'] dict
        and file['storage'] will be set to None.

        Args:
            files: List of dicts containing file data
            allow_create: Create FileStorages that don't exist.
            raise_exception: Raise exception on error. Otherwise add errors to dict.
            remove_filestorage_fields: When enabled, remove storage_service and
                other FileStorage fields from the files.

        Returns:
            files: Modified file data.
        """

        files_by_key = self._group_files_by_key(files, raise_exception=raise_exception)
        storages_by_key = self._get_existing_filestorages_by_key(files_by_key)
        storages_by_key, errors_by_key = self._create_missing_filestorages(
            files_by_key,
            storages_by_key,
            allow_create=allow_create,
            raise_exception=raise_exception,
        )

        # Assign FileStorage instances to data
        for key, key_files in files_by_key.items():
            for f in key_files:
                if f.get("errors"):
                    f["storage"] = None
                    continue
                if remove_filestorage_fields:
                    f.pop("storage_service", None)
                    for field in self.model.all_extra_fields:
                        f.pop(field, None)
                f["storage"] = storages_by_key.get(key)
                if errors := errors_by_key.get(key):
                    f.setdefault("errors", {}).update(errors)
        return files

assign_to_file_data(files, allow_create=False, raise_exception=True, remove_filestorage_fields=False)

Retrieve FileStorage instances and assign to file data.

Retrieves (or optionally creates) a FileStorage instance for each file in files list. The instance is assigned to file['storage'].

Files are validated for required FileStorage fields based on file['storage_service']. When raise_exception is False and an error occurs, the error will be added to file['errors'] dict and file['storage'] will be set to None.

Parameters:

Name	Type	Description	Default
files	List[dict]	List of dicts containing file data	required
allow_create		Create FileStorages that don't exist.	False
raise_exception		Raise exception on error. Otherwise add errors to dict.	True
remove_filestorage_fields		When enabled, remove storage_service and other FileStorage fields from the files.	False

Returns:

Name	Type	Description
files	List[dict]	Modified file data.

Source code in src/apps/files/models/file_storage.py

def assign_to_file_data(
    self,
    files: List[dict],
    allow_create=False,
    raise_exception=True,
    remove_filestorage_fields=False,
) -> List[dict]:
    """
    Retrieve FileStorage instances and assign to file data.

    Retrieves (or optionally creates) a FileStorage instance for
    each file in files list. The instance is assigned to
    file['storage'].

    Files are validated for required FileStorage fields based on
    file['storage_service']. When raise_exception is False and
    an error occurs, the error will be added to file['errors'] dict
    and file['storage'] will be set to None.

    Args:
        files: List of dicts containing file data
        allow_create: Create FileStorages that don't exist.
        raise_exception: Raise exception on error. Otherwise add errors to dict.
        remove_filestorage_fields: When enabled, remove storage_service and
            other FileStorage fields from the files.

    Returns:
        files: Modified file data.
    """

    files_by_key = self._group_files_by_key(files, raise_exception=raise_exception)
    storages_by_key = self._get_existing_filestorages_by_key(files_by_key)
    storages_by_key, errors_by_key = self._create_missing_filestorages(
        files_by_key,
        storages_by_key,
        allow_create=allow_create,
        raise_exception=raise_exception,
    )

    # Assign FileStorage instances to data
    for key, key_files in files_by_key.items():
        for f in key_files:
            if f.get("errors"):
                f["storage"] = None
                continue
            if remove_filestorage_fields:
                f.pop("storage_service", None)
                for field in self.model.all_extra_fields:
                    f.pop(field, None)
            f["storage"] = storages_by_key.get(key)
            if errors := errors_by_key.get(key):
                f.setdefault("errors", {}).update(errors)
    return files

get_for_object(object)

Get FileStorage for a dict object.

Validates input object and returns single FileStorage instance based on its storage_service value and FileStorage-specific extra values.

Source code in src/apps/files/models/file_storage.py

def get_for_object(self, object: dict) -> models.Model:
    """Get FileStorage for a dict object.

    Validates input object and returns single FileStorage instance
    based on its storage_service value and FileStorage-specific extra values.
    """
    self.model.validate_object(object)
    try:
        return self.get(
            csc_project=object["csc_project"],
            storage_service=object["storage_service"],
        )
    except self.model.DoesNotExist:
        raise exceptions.NotFound()

SoftDeletableFileStorageManager

Bases: FileStorageManagerMixin, SoftDeletableManager

Source code in src/apps/files/models/file_storage.py

class SoftDeletableFileStorageManager(FileStorageManagerMixin, SoftDeletableManager):
    pass

FileStorageManager

Bases: FileStorageManagerMixin, Manager

Source code in src/apps/files/models/file_storage.py

class FileStorageManager(FileStorageManagerMixin, models.Manager):
    pass

FileStorage

Bases: ProxyBasePolymorphicModel, AbstractBaseModel

FileStorage respresents a collection of files in a storage service.

Source code in src/apps/files/models/file_storage.py

class FileStorage(ProxyBasePolymorphicModel, AbstractBaseModel):
    """FileStorage respresents a collection of files in a storage service."""

    proxy_lookup_field = "storage_service"

    proxy_mapping = settings.STORAGE_SERVICE_FILE_STORAGES

    # Extended SoftDeletableModel managers
    objects = SoftDeletableFileStorageManager(_emit_deprecation_warnings=True)
    available_objects = SoftDeletableFileStorageManager()
    all_objects = FileStorageManager()

    # Fields common for all FileStorage types
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    storage_service = models.CharField(max_length=64, choices=STORAGE_SERVICE_CHOICES)

    # Extra identification fields used only by specific FileStorage types
    csc_project = models.CharField(max_length=200, null=True, blank=True)
    all_extra_fields = ["csc_project"]

    # A key should uniquely identify a FileStorage instance
    key_type = namedtuple("FileStorageKey", ["storage_service", *all_extra_fields])

    # Set supported extra fields for model. Extra fields not listed here should be null.
    optional_extra_fields = set()
    required_extra_fields = set()

    # File fields that are unique by FileStorage or file service, validation in Python.
    # Any null values are exempt from the uniqueness check. File path has its own validation
    # and is not listed here.
    unique_file_fields_per_storage = []
    unique_file_fields_per_storage_service = ["storage_identifier"]

    # File fields required by FileStorage that are normally optional, e.g. storage_identifier
    required_file_fields = set()

    class Meta(AbstractBaseModel.Meta):
        unique_together = [("csc_project", "storage_service")]

    @classmethod
    @functools.lru_cache
    def get_allowed_extra_fields(cls):
        return {*cls.optional_extra_fields, *cls.required_extra_fields}

    @classmethod
    def get_key(cls, object) -> tuple:
        """Uniquely identify a FileStorage instance based on its field values."""
        return cls.key_type(*(get_attr_or_item(object, field) for field in cls.key_type._fields))

    @classmethod
    def remove_unsupported_extra_fields(cls, data: dict) -> dict:
        """Remove disallowed FileService fields from data.

        E.g. remove project from dict if the current
        FileStorage does not support the project field."""
        allowed_extra_fields = cls.get_allowed_extra_fields()
        for field in cls.all_extra_fields:
            if field not in allowed_extra_fields:
                data.pop(field, None)
        return data

    def check_user_can_access(self, user: MetaxUser):
        """Check user can access files in the FileSorage.

        Subclasses should raise a ValidationError if user is not allowed to
        add files from the FileStorage to a FileSet."""

    @classmethod
    def validate_object(cls, object: Union["FileStorage", dict]):
        """Make sure object has required file storage fields as attributes or items.

        Useful for:
        * Checking a FileStorage object is valid and can be saved.
        * Checking that file dict values correspond to a valid FileStorage.
        """
        if isinstance(object, dict) and object.get("errors", {}).get("id"):
            # File not found, skip remaining validation here
            # to avoid misleading errors in fields that would be filled
            # automatically for existing fields
            return

        storage_service = get_attr_or_item(object, "storage_service")
        if not storage_service:
            raise ValidationError({"storage_service": _("Value is required.")})
        if storage_service not in settings.STORAGE_SERVICE_FILE_STORAGES:
            raise ValidationError({"storage_service": _("Invalid storage_service value.")})

        proxy_cls = cls.get_proxy_model(storage_service)
        for field in cls.all_extra_fields:
            required = field in proxy_cls.required_extra_fields
            forbidden = field not in proxy_cls.get_allowed_extra_fields()
            value = get_attr_or_item(object, field)
            error_msg = None
            if required and value is None:
                error_msg = _("Field is required for storage_service='{location}'.")
            elif forbidden and value is not None:
                error_msg = _("Field is not allowed for storage_service='{location}'.")
            if error_msg:
                raise ValidationError({field: error_msg.format(location=storage_service)})

    def __repr__(self):
        relevant_fields = {
            "storage_service",
            *self.required_extra_fields,
            *self.optional_extra_fields,
        }
        values = ", ".join(
            [f"{k}={v}" for k, v in self.key._asdict().items() if k in relevant_fields]
        )
        return f"<{self.__class__.__name__}: {values}>"

    def get_directory_paths(self, file_set=None, include_removed=False) -> Set[str]:
        """Get directory paths used in the storage as a set.

        If dataset is supplied, return only directories belonging to dataset.
        Otherwise all directories are returned."""
        qs = self.files
        if include_removed:
            qs = self.files(manager="all_objects")
        if file_set:
            qs = qs.filter(file_sets=file_set)
        file_pathnames = (
            qs.values_list("directory_path", flat=True)
            .order_by("directory_path")
            .distinct("directory_path")
        )
        all_paths = set(file_pathnames)

        # Add intermediate directories that don't have files directly but in subdirs.
        last_part = re.compile("/[^/]+/$")  # matches e.g. `/subdir/` for `/dir/subdir/`
        for path in file_pathnames:
            # Remove last path part and add to set until encountering path already in set.
            path = last_part.sub("/", path, count=1)
            while path not in all_paths:
                all_paths.add(path)
                path = last_part.sub("/", path, count=1)
        return all_paths

    def save(self, *args, **kwargs):
        try:
            self.validate_object(self)
        except ValidationError as e:
            raise ValueError(e.detail)
        return super().save(*args, **kwargs)

    @property
    def key(self) -> tuple:
        return self.get_key(self)

    @property
    def params_dict(self):
        return self.remove_unsupported_extra_fields(self.key._asdict())

    # Validation for file data conflics

    @classmethod
    def _group_file_data_by_storage(cls, files: List[dict]) -> Dict["FileStorage", dict]:
        files_by_storage = {}
        for f in files:
            if f["storage"] is not None:
                files_by_storage.setdefault(f["storage"], []).append(f)
        return files_by_storage

    @classmethod
    def _group_file_data_by_storage_service(cls, files: List[dict]) -> Dict[str, dict]:
        files_by_service = {}
        for f in files:
            if f["storage"] is not None:
                files_by_service.setdefault(f["storage"].storage_service, []).append(f)
        return files_by_service

    @classmethod
    def _check_conflicts_within_new(
        cls, files: List[dict], field, raise_exception=True
    ) -> List[dict]:
        """Check new file data for duplicate values."""
        new_values = set()
        for f in files:
            value = f.get(field)
            if value is not None:
                if value in new_values:
                    errors = {field: _("Duplicate value in request.")}
                    if raise_exception:
                        raise ValidationError(errors)
                    else:
                        f.setdefault("errors", {}).update(errors)
                else:
                    new_values.add(value)
        return files

    @classmethod
    def _check_conflicts_with_existing(
        cls, files: List[dict], queryset, field, raise_exception=True
    ) -> List[dict]:
        """Check file data for conflicts with existing data."""
        new_values = [f[field] for f in files if f.get(field) is not None]
        conflicts = queryset.filter(**{f"{field}__in": new_values}).values(field, "id")
        if conflicts:
            conflicting_id_by_value = {c[field]: c["id"] for c in conflicts}
            for f in files:
                errors = None
                if f.get(field) in conflicting_id_by_value:
                    errors = {
                        field: _("A file with the same value already exists, id='{fid}'.").format(
                            fid=conflicting_id_by_value[f[field]]
                        )
                    }
                if errors:
                    if raise_exception:
                        raise ValidationError(errors)
                    else:
                        f.setdefault("errors", {}).update(errors)
        return files

    @classmethod
    def _check_conflicts(cls, files: List[dict], queryset, field, raise_exception):
        """Check conflicts in file data."""
        files = cls._check_conflicts_within_new(files, field, raise_exception)
        files = cls._check_conflicts_with_existing(files, queryset, field, raise_exception)

    @classmethod
    def _check_file_path_conflicts(cls, files: List[dict], raise_exception):
        """Check pathname conflicts."""
        new_files = [f for f in files if ("id" not in f)]
        new_files_by_storage = cls._group_file_data_by_storage(new_files)
        for storage, new_storage_files in new_files_by_storage.items():
            new_values = [f["pathname"] for f in files if f.get("pathname")]
            queryset = storage.files.filter(
                # prefilter results before doing a more expensive exact match with Concat
                directory_path__in=set(p.rsplit("/", 1)[0] + "/" for p in new_values),
                filename__in=set(p.rsplit("/", 1)[1] for p in new_values),
            ).annotate(pathname=Concat("directory_path", "filename"))
            cls._check_conflicts(
                new_storage_files,
                queryset=queryset,
                field="pathname",
                raise_exception=raise_exception,
            )
        return files

    @classmethod
    def _check_storage_value_conflicts(cls, files: List[dict], raise_exception):
        """Check conflicts with fields that are unique per file storage."""
        new_files = [f for f in files if ("id" not in f)]
        new_files_by_storage = cls._group_file_data_by_storage(new_files)
        for storage, new_storage_files in new_files_by_storage.items():
            queryset = storage.files
            for field in storage.unique_file_fields_per_storage:
                cls._check_conflicts(
                    new_storage_files,
                    queryset=queryset,
                    field=field,
                    raise_exception=raise_exception,
                )
        return files

    @classmethod
    def _check_file_service_value_conflicts(cls, files: List[dict], raise_exception):
        """Check conflicts with fields that are unique per file service."""
        from apps.files.models import File

        new_files = [f for f in files if ("id" not in f) and (f["storage"] is not None)]
        new_files_by_service = cls._group_file_data_by_storage_service(new_files)
        for storage_service, new_service_files in new_files_by_service.items():
            proxy_model = cls.get_proxy_model(storage_service)
            queryset = File.objects.filter(storage__storage_service=storage_service)
            for field in proxy_model.unique_file_fields_per_storage_service:
                cls._check_conflicts(
                    new_service_files,
                    queryset=queryset,
                    field=field,
                    raise_exception=raise_exception,
                )
        return files

    @classmethod
    def check_file_data_conflicts(cls, files: List[dict], raise_exception=True) -> List[dict]:
        """Check values that should be unique for FileStorage or file service.

        Input Files are validated for required FileStorage fields based on
        storage_service value. When raise_exception is False and
        an error occurs, the error will be added to file['errors'] dict.

        Args:
            files: List of dicts containing file data
            raise_exception: Raise exception on error. Otherwise add errors to dict.

        Returns:
            files: File data with included errors.
        """
        files = cls._check_file_path_conflicts(files, raise_exception=raise_exception)
        files = cls._check_storage_value_conflicts(files, raise_exception=raise_exception)
        files = cls._check_file_service_value_conflicts(files, raise_exception=raise_exception)
        return files

    @classmethod
    def check_required_file_fields(cls, files: List[dict], raise_exception=True) -> List[dict]:
        """Check new file data for missing field values that are required by the FileStorage."""
        new_files = [f for f in files if ("id" not in f)]
        new_files_by_storage = cls._group_file_data_by_storage(new_files)
        for storage, storage_files in new_files_by_storage.items():
            required_fields = list(storage.required_file_fields)
            for file in storage_files:
                missing_fields = {field for field in required_fields if not file.get(field)}
                if missing_fields:
                    errors = {
                        field: _("Field is required for storage_service '{}'").format(
                            storage.storage_service
                        )
                        for field in missing_fields
                    }
                    if raise_exception:
                        raise ValidationError(errors)
                    file.setdefault("errors", {}).update(errors)

        return files

    @classmethod
    def get_or_create_from_legacy(cls, legacy_file: dict) -> "FileStorage":
        legacy_storage = legacy_file["file_storage"]["identifier"]
        csc_project = legacy_file["project_identifier"]
        storage_service = settings.LEGACY_FILE_STORAGE_TO_V3_STORAGE_SERVICE.get(legacy_storage)
        if not storage_service:
            raise ValueError(f"Unknown legacy storage service: {legacy_storage}")

        if storage_service not in settings.STORAGE_SERVICE_FILE_STORAGES:
            raise ValueError({"storage_service": f"Unknown storage service: '{storage_service}'"})
        storage, _created = cls.available_objects.get_or_create(
            csc_project=csc_project,
            storage_service=storage_service,
        )
        return storage

check_file_data_conflicts(files, raise_exception=True) classmethod

Check values that should be unique for FileStorage or file service.

Input Files are validated for required FileStorage fields based on storage_service value. When raise_exception is False and an error occurs, the error will be added to file['errors'] dict.

Parameters:

Name	Type	Description	Default
files	List[dict]	List of dicts containing file data	required
raise_exception		Raise exception on error. Otherwise add errors to dict.	True

Returns:

Name	Type	Description
files	List[dict]	File data with included errors.

Source code in src/apps/files/models/file_storage.py

@classmethod
def check_file_data_conflicts(cls, files: List[dict], raise_exception=True) -> List[dict]:
    """Check values that should be unique for FileStorage or file service.

    Input Files are validated for required FileStorage fields based on
    storage_service value. When raise_exception is False and
    an error occurs, the error will be added to file['errors'] dict.

    Args:
        files: List of dicts containing file data
        raise_exception: Raise exception on error. Otherwise add errors to dict.

    Returns:
        files: File data with included errors.
    """
    files = cls._check_file_path_conflicts(files, raise_exception=raise_exception)
    files = cls._check_storage_value_conflicts(files, raise_exception=raise_exception)
    files = cls._check_file_service_value_conflicts(files, raise_exception=raise_exception)
    return files

check_required_file_fields(files, raise_exception=True) classmethod

Check new file data for missing field values that are required by the FileStorage.

Source code in src/apps/files/models/file_storage.py

@classmethod
def check_required_file_fields(cls, files: List[dict], raise_exception=True) -> List[dict]:
    """Check new file data for missing field values that are required by the FileStorage."""
    new_files = [f for f in files if ("id" not in f)]
    new_files_by_storage = cls._group_file_data_by_storage(new_files)
    for storage, storage_files in new_files_by_storage.items():
        required_fields = list(storage.required_file_fields)
        for file in storage_files:
            missing_fields = {field for field in required_fields if not file.get(field)}
            if missing_fields:
                errors = {
                    field: _("Field is required for storage_service '{}'").format(
                        storage.storage_service
                    )
                    for field in missing_fields
                }
                if raise_exception:
                    raise ValidationError(errors)
                file.setdefault("errors", {}).update(errors)

    return files

check_user_can_access(user)

Check user can access files in the FileSorage.

Subclasses should raise a ValidationError if user is not allowed to add files from the FileStorage to a FileSet.

Source code in src/apps/files/models/file_storage.py

def check_user_can_access(self, user: MetaxUser):
    """Check user can access files in the FileSorage.

    Subclasses should raise a ValidationError if user is not allowed to
    add files from the FileStorage to a FileSet."""

get_directory_paths(file_set=None, include_removed=False)

Get directory paths used in the storage as a set.

If dataset is supplied, return only directories belonging to dataset. Otherwise all directories are returned.

Source code in src/apps/files/models/file_storage.py

def get_directory_paths(self, file_set=None, include_removed=False) -> Set[str]:
    """Get directory paths used in the storage as a set.

    If dataset is supplied, return only directories belonging to dataset.
    Otherwise all directories are returned."""
    qs = self.files
    if include_removed:
        qs = self.files(manager="all_objects")
    if file_set:
        qs = qs.filter(file_sets=file_set)
    file_pathnames = (
        qs.values_list("directory_path", flat=True)
        .order_by("directory_path")
        .distinct("directory_path")
    )
    all_paths = set(file_pathnames)

    # Add intermediate directories that don't have files directly but in subdirs.
    last_part = re.compile("/[^/]+/$")  # matches e.g. `/subdir/` for `/dir/subdir/`
    for path in file_pathnames:
        # Remove last path part and add to set until encountering path already in set.
        path = last_part.sub("/", path, count=1)
        while path not in all_paths:
            all_paths.add(path)
            path = last_part.sub("/", path, count=1)
    return all_paths

get_key(object) classmethod

Uniquely identify a FileStorage instance based on its field values.

Source code in src/apps/files/models/file_storage.py

@classmethod
def get_key(cls, object) -> tuple:
    """Uniquely identify a FileStorage instance based on its field values."""
    return cls.key_type(*(get_attr_or_item(object, field) for field in cls.key_type._fields))

remove_unsupported_extra_fields(data) classmethod

Remove disallowed FileService fields from data.

E.g. remove project from dict if the current FileStorage does not support the project field.

Source code in src/apps/files/models/file_storage.py

@classmethod
def remove_unsupported_extra_fields(cls, data: dict) -> dict:
    """Remove disallowed FileService fields from data.

    E.g. remove project from dict if the current
    FileStorage does not support the project field."""
    allowed_extra_fields = cls.get_allowed_extra_fields()
    for field in cls.all_extra_fields:
        if field not in allowed_extra_fields:
            data.pop(field, None)
    return data

validate_object(object) classmethod

Make sure object has required file storage fields as attributes or items.

Useful for: * Checking a FileStorage object is valid and can be saved. * Checking that file dict values correspond to a valid FileStorage.

Source code in src/apps/files/models/file_storage.py

@classmethod
def validate_object(cls, object: Union["FileStorage", dict]):
    """Make sure object has required file storage fields as attributes or items.

    Useful for:
    * Checking a FileStorage object is valid and can be saved.
    * Checking that file dict values correspond to a valid FileStorage.
    """
    if isinstance(object, dict) and object.get("errors", {}).get("id"):
        # File not found, skip remaining validation here
        # to avoid misleading errors in fields that would be filled
        # automatically for existing fields
        return

    storage_service = get_attr_or_item(object, "storage_service")
    if not storage_service:
        raise ValidationError({"storage_service": _("Value is required.")})
    if storage_service not in settings.STORAGE_SERVICE_FILE_STORAGES:
        raise ValidationError({"storage_service": _("Invalid storage_service value.")})

    proxy_cls = cls.get_proxy_model(storage_service)
    for field in cls.all_extra_fields:
        required = field in proxy_cls.required_extra_fields
        forbidden = field not in proxy_cls.get_allowed_extra_fields()
        value = get_attr_or_item(object, field)
        error_msg = None
        if required and value is None:
            error_msg = _("Field is required for storage_service='{location}'.")
        elif forbidden and value is not None:
            error_msg = _("Field is not allowed for storage_service='{location}'.")
        if error_msg:
            raise ValidationError({field: error_msg.format(location=storage_service)})

BasicFileStorage

Bases: FileStorage

FileStorage that does not support any extra fields.

Source code in src/apps/files/models/file_storage.py

class BasicFileStorage(FileStorage):
    """FileStorage that does not support any extra fields."""

    class Meta:
        proxy = True

ProjectFileStorage

Bases: FileStorage

FileStorage that requires csc_project to be set.

Source code in src/apps/files/models/file_storage.py

class ProjectFileStorage(FileStorage):
    """FileStorage that requires csc_project to be set."""

    required_extra_fields = {"csc_project"}

    def check_user_can_access(self, user: MetaxUser):
        """Check user is member of csc_project or belongs to privileged group."""
        super().check_user_can_access(user)
        service_groups = settings.PROJECT_STORAGE_SERVICE_USER_GROUPS
        if not (
            user.is_superuser
            or self.csc_project in user.csc_projects
            or user.groups.filter(name__in=service_groups).exists()
        ):
            raise ValidationError(
                {"csc_project": f"User does not have access to project {self.csc_project}"}
            )

    class Meta:
        proxy = True

check_user_can_access(user)

Check user is member of csc_project or belongs to privileged group.

Source code in src/apps/files/models/file_storage.py

def check_user_can_access(self, user: MetaxUser):
    """Check user is member of csc_project or belongs to privileged group."""
    super().check_user_can_access(user)
    service_groups = settings.PROJECT_STORAGE_SERVICE_USER_GROUPS
    if not (
        user.is_superuser
        or self.csc_project in user.csc_projects
        or user.groups.filter(name__in=service_groups).exists()
    ):
        raise ValidationError(
            {"csc_project": f"User does not have access to project {self.csc_project}"}
        )

IDAFileStorage

Bases: ProjectFileStorage

Source code in src/apps/files/models/file_storage.py

class IDAFileStorage(ProjectFileStorage):
    required_file_fields = {"storage_identifier"}

    class Meta:
        proxy = True