Skip to content

metadata_provider_serializer#

UsernameField#

Bases: CharField

CharField that returns username when serializing a user.

Normally a SlugRelatedField could be used instead but this way doesn't require the user to already exist.

Source code in src/apps/core/serializers/metadata_provider_serializer.py
class UsernameField(serializers.CharField):
    """CharField that returns username when serializing a user.

    Normally a SlugRelatedField could be used instead but
    this way doesn't require the user to already exist.
    """

    def to_representation(self, value):
        if isinstance(value, MetaxUser):
            return value.username
        return value

MetadataProviderModelSerializer#

Bases: StrictSerializer, AbstractDatasetModelSerializer

Metadata provider handling logic.

Creates or retrieves MetadataProvider object based on request data. Does not alter existing objects.

Call save() with serializer._validated_data={} to use default provider. Defaults are: - If instance exists, use provider values from instance - If instance does not exist, use provider values from request user

Source code in src/apps/core/serializers/metadata_provider_serializer.py
class MetadataProviderModelSerializer(StrictSerializer, AbstractDatasetModelSerializer):
    """Metadata provider handling logic.

    Creates or retrieves MetadataProvider object based on request data.
    Does not alter existing objects.

    Call `save()` with `serializer._validated_data={}` to use default provider. Defaults are:
    - If instance exists, use provider values from instance
    - If instance does not exist, use provider values from request user
    """

    user = UsernameField()

    class Meta:
        model = MetadataProvider
        fields = ("id", "user", "organization")

    def is_custom_value_allowed(self):
        """End-users should not be allowed to use custom values."""
        if self.context.get("migrating"):
            return True
        request = self.context["request"]
        return DatasetAccessPolicy().query_object_permission(  # note that instance may be null
            user=request.user, object=self.instance, action="<op:custom_metadata_owner>"
        )

    def save(self, **kwargs):
        """Save with support for custom values.

        Call save with an empty dict as self._validated_data
        to use request user."""
        data = self._validated_data

        # Default to existing values or values from authenticated user
        if data and self.is_custom_value_allowed():
            ctx_data = data
        elif self.instance:
            ctx_data = {
                "user": self.instance.user.username,
                "organization": self.instance.organization,
            }
        else:
            user = self.context["request"].user
            ctx_data = {
                "user": user.username,
                "organization": user.organization or user.username,
            }

        if not data:
            # Data is empty, use values from request user
            data = ctx_data
        else:
            # Data has values but they should match the request user
            errors = {}
            if data["user"] != ctx_data["user"]:
                errors["user"] = _("Not allowed to change value. Expected '{}'.").format(
                    ctx_data["user"]
                )
            if data["organization"] != ctx_data["organization"]:
                errors["organization"] = _(
                    _("Not allowed to change value. Expected '{}'.")
                ).format(ctx_data["organization"])
            if errors:
                raise serializers.ValidationError(errors)
        return super().save(**data)

    def get_or_create_provider(self, validated_data):
        """Reuse existing metadata provider when possible."""
        user, created = get_user_model().objects.get_or_create(username=validated_data["user"])
        organization = validated_data["organization"]
        provider = MetadataProvider.objects.filter(user=user, organization=organization).first()
        if provider:
            return provider
        return MetadataProvider.objects.create(user=user, organization=organization)

    def create(self, validated_data):
        return self.get_or_create_provider(validated_data)

    def update(self, instance, validated_data):
        return self.get_or_create_provider(validated_data)

get_or_create_provider(validated_data) #

Reuse existing metadata provider when possible.

Source code in src/apps/core/serializers/metadata_provider_serializer.py
def get_or_create_provider(self, validated_data):
    """Reuse existing metadata provider when possible."""
    user, created = get_user_model().objects.get_or_create(username=validated_data["user"])
    organization = validated_data["organization"]
    provider = MetadataProvider.objects.filter(user=user, organization=organization).first()
    if provider:
        return provider
    return MetadataProvider.objects.create(user=user, organization=organization)

is_custom_value_allowed() #

End-users should not be allowed to use custom values.

Source code in src/apps/core/serializers/metadata_provider_serializer.py
def is_custom_value_allowed(self):
    """End-users should not be allowed to use custom values."""
    if self.context.get("migrating"):
        return True
    request = self.context["request"]
    return DatasetAccessPolicy().query_object_permission(  # note that instance may be null
        user=request.user, object=self.instance, action="<op:custom_metadata_owner>"
    )

save(**kwargs) #

Save with support for custom values.

Call save with an empty dict as self._validated_data to use request user.

Source code in src/apps/core/serializers/metadata_provider_serializer.py
def save(self, **kwargs):
    """Save with support for custom values.

    Call save with an empty dict as self._validated_data
    to use request user."""
    data = self._validated_data

    # Default to existing values or values from authenticated user
    if data and self.is_custom_value_allowed():
        ctx_data = data
    elif self.instance:
        ctx_data = {
            "user": self.instance.user.username,
            "organization": self.instance.organization,
        }
    else:
        user = self.context["request"].user
        ctx_data = {
            "user": user.username,
            "organization": user.organization or user.username,
        }

    if not data:
        # Data is empty, use values from request user
        data = ctx_data
    else:
        # Data has values but they should match the request user
        errors = {}
        if data["user"] != ctx_data["user"]:
            errors["user"] = _("Not allowed to change value. Expected '{}'.").format(
                ctx_data["user"]
            )
        if data["organization"] != ctx_data["organization"]:
            errors["organization"] = _(
                _("Not allowed to change value. Expected '{}'.")
            ).format(ctx_data["organization"])
        if errors:
            raise serializers.ValidationError(errors)
    return super().save(**data)