Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if data['file'].size != int(data['size']):
raise serializers.ValidationError(_("The size did not match actual size of file."))
else:
data['size'] = data['file'].size
for algorithm in hashlib.algorithms_guaranteed:
if algorithm in models.Artifact.DIGEST_FIELDS:
digest = data['file'].hashers[algorithm].hexdigest()
if algorithm in data and digest != data[algorithm]:
raise serializers.ValidationError(_("The %s checksum did not match.")
% algorithm)
else:
data[algorithm] = digest
if algorithm in UNIQUE_ALGORITHMS:
validator = UniqueValidator(models.Artifact.objects.all(),
message=_("{0} checksum must be "
"unique.").format(algorithm))
validator.field_name = algorithm
validator.instance = None
validator(digest)
return data
- define its own `Meta` class should:
- specify plugin content model
- extend `fields` with plugin-specific ones
"""
class Meta:
model = Artifact
fields = ['md5', 'sha1', 'sha224', 'sha256', 'sha384', 'sha512']
class ArtifactViewSet(NamedModelViewSet,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
mixins.DestroyModelMixin):
endpoint_name = 'artifacts'
queryset = Artifact.objects.all()
serializer_class = ArtifactSerializer
filterset_class = ArtifactFilter
parser_classes = (MultiPartParser, FormParser)
swagger_schema = ArtifactSchema
def destroy(self, request, pk):
"""
Remove Artifact only if it is not associated with any Content.
"""
try:
return super().destroy(request, pk)
except models.ProtectedError:
msg = _('The Artifact cannot be deleted because it is associated with Content.')
data = {'detail': msg}
return Response(data, status=status.HTTP_409_CONFLICT)
class DownloadCatalogSerializer(ModelSerializer):
_href = serializers.HyperlinkedIdentityField(
view_name='downloadcatalogs-detail',
)
url = serializers.CharField(
help_text=_("The URL used to download the related artifact."),
allow_blank=True, read_only=True,
)
artifact = serializers.HyperlinkedRelatedField(
help_text=_("The artifact that is expected to be present at url"),
queryset=models.Artifact.objects.all(),
view_name="artifacts-detail"
)
importer = DetailNestedHyperlinkedRelatedField(
parent_lookup_kwargs={'repository_name': 'repository__name'},
queryset=models.Importer.objects.all(),
help_text=_("The importer that contains the configuration necessary to access url."),
lookup_field='name'
)
class Meta:
model = models.DownloadCatalog
fields = ModelSerializer.Meta.fields + ("artifact", "importer", "url",)
class ContentRelatedField(DetailRelatedField):
"""
Serializer Field for use when relating to Content Detail Models
"""
queryset = models.Content.objects.all()
class SingleContentArtifactField(RelatedField):
"""
A serializer field for the '_artifacts' ManyToManyField on the Content model (single-artifact).
"""
lookup_field = 'pk'
view_name = 'artifacts-detail'
queryset = models.Artifact.objects.all()
allow_null = True
def get_attribute(self, instance):
"""
Returns the field from the instance that should be serialized using this serializer field.
This serializer looks up the list of artifacts and returns only one, if any exist. If more
than one exist, it throws and exception because this serializer is being used in an
improper context.
Args:
instance (:class:`pulpcore.app.models.Content`): An instance of Content being
serialized.
Returns:
A single Artifact model related to the instance of Content.
class DownloadCatalogSerializer(ModelSerializer):
_href = serializers.HyperlinkedIdentityField(
view_name='downloadcatalogs-detail',
)
url = serializers.CharField(
help_text=_("The URL used to download the related artifact."),
allow_blank=True, read_only=True,
)
artifact = serializers.HyperlinkedRelatedField(
help_text=_("The artifact that is expected to be present at url"),
queryset=models.Artifact.objects.all(),
view_name="artifacts-detail"
)
importer = DetailNestedHyperlinkedRelatedField(
parent_lookup_kwargs={'repository_name': 'repository__name'},
queryset=models.Importer.objects.all(),
help_text=_("The importer that contains the configuration necessary to access url."),
lookup_field='name'
)
class Meta:
model = models.DownloadCatalog
fields = ModelSerializer.Meta.fields + ("artifact", "importer", "url",)
"sha224",
"sha256",
"sha384",
"sha512",
}
class ArtifactViewSet(
NamedModelViewSet,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
mixins.DestroyModelMixin,
):
endpoint_name = "artifacts"
queryset = Artifact.objects.all()
serializer_class = ArtifactSerializer
filterset_class = ArtifactFilter
def destroy(self, request, pk):
"""
Remove Artifact only if it is not associated with any Content.
"""
try:
return super().destroy(request, pk)
except models.ProtectedError:
msg = _("The Artifact cannot be deleted because it is associated with Content.")
data = {"detail": msg}
return Response(data, status=status.HTTP_409_CONFLICT)
class ContentFilter(BaseFilterSet):
else:
data["size"] = data["file"].size
for algorithm in hashlib.algorithms_guaranteed:
if algorithm in models.Artifact.DIGEST_FIELDS:
digest = data["file"].hashers[algorithm].hexdigest()
if algorithm in data and digest != data[algorithm]:
raise serializers.ValidationError(
_("The %s checksum did not match.") % algorithm
)
else:
data[algorithm] = digest
if algorithm in UNIQUE_ALGORITHMS:
validator = UniqueValidator(
models.Artifact.objects.all(),
message=_("{0} checksum must be " "unique.").format(algorithm),
)
validator.field_name = algorithm
validator.instance = None
validator(digest)
return data