Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
model = models.TaskGroup
fields = (
"pulp_href",
"description",
"all_tasks_dispatched",
"waiting",
"skipped",
"running",
"completed",
"canceled",
"failed",
"group_progress_reports",
)
class TaskCancelSerializer(ModelSerializer):
state = serializers.CharField(
help_text=_("The desired state of the task. Only 'canceled' is accepted."),
)
class Meta:
model = models.Task
fields = ("state",)
class ContentAppStatusSerializer(ModelSerializer):
name = serializers.CharField(help_text=_("The name of the worker."), read_only=True)
last_heartbeat = serializers.DateTimeField(
help_text=_("Timestamp of the last time the worker talked to the service."), read_only=True
)
class Meta:
),
source="*",
read_only=True,
)
class Meta:
model = models.RepositoryVersion
fields = ModelSerializer.Meta.fields + (
"pulp_href",
"number",
"base_version",
"content_summary",
)
class RepositoryAddRemoveContentSerializer(ModelSerializer, NestedHyperlinkedModelSerializer):
add_content_units = serializers.ListField(
help_text=_(
"A list of content units to add to a new repository version. This content is "
"added after remove_content_units are removed."
),
required=False,
)
remove_content_units = serializers.ListField(
help_text=_(
"A list of content units to remove from the latest repository version. "
"You may also specify '*' as an entry to remove all content. This content is "
"removed before add_content_units are added."
),
required=False,
)
base_version = RepositoryVersionRelatedField(
def import_serializers(self):
# circular import avoidance
from pulpcore.app.serializers import ModelSerializer
self.named_serializers = {}
if module_has_submodule(self.module, SERIALIZERS_MODULE_NAME):
# import the serializers module and track any discovered serializers
serializers_module_name = '{name}.{module}'.format(
name=self.name, module=SERIALIZERS_MODULE_NAME)
self.serializers_module = import_module(serializers_module_name)
for objname in dir(self.serializers_module):
obj = getattr(self.serializers_module, objname)
try:
# Any subclass of ModelSerializer that isn't itself ModelSerializer
# gets registered in the named_serializers registry.
if obj is not ModelSerializer and issubclass(obj, ModelSerializer):
self.named_serializers[objname] = obj
except TypeError:
# obj isn't a class, issubclass exploded but obj can be safely filtered out
continue
help_text=_('Resources created by this task.'),
many=True,
read_only=True
)
class Meta:
model = models.Task
fields = ModelSerializer.Meta.fields + ('state', 'started_at', 'finished_at',
'non_fatal_errors', 'error', 'worker', 'parent',
'spawned_tasks', 'progress_reports',
'created_resources')
minimal_fields = ModelSerializer.Meta.fields + ('state', 'started_at', 'finished_at',
'worker', 'parent')
class WorkerSerializer(ModelSerializer):
_href = serializers.HyperlinkedIdentityField(view_name='workers-detail')
name = serializers.CharField(
help_text=_('The name of the worker.'),
read_only=True
)
last_heartbeat = serializers.DateTimeField(
help_text=_('Timestamp of the last time the worker talked to the service.'),
read_only=True
)
online = serializers.BooleanField(
help_text=_('True if the worker is considered online, otherwise False'),
read_only=True
)
missing = serializers.BooleanField(
help_text=_('True if the worker is considerd missing, otherwise False'),
required=False,
allow_null=True,
)
exported_resources = ExportedResourcesField(
help_text=_("Resources that were exported."), source="*", read_only=True,
)
params = serializers.JSONField(
help_text=_("Any additional parameters that were used to create the export."),
read_only=True,
)
class Meta:
model = models.Export
fields = ModelSerializer.Meta.fields + ("task", "exported_resources", "params")
class PulpExportSerializer(ExportSerializer):
"""
Serializer for PulpExports.
"""
output_file_info = serializers.JSONField(
help_text=_("Dictionary of filename: sha256hash entries for export-output-file(s)"),
read_only=True,
)
dry_run = serializers.BooleanField(
help_text=_("Generate report on what would be exported and disk-space required."),
default=False,
required=False,
def import_serializers(self):
# circular import avoidance
from pulpcore.app.serializers import ModelSerializer
self.named_serializers = {}
if module_has_submodule(self.module, SERIALIZERS_MODULE_NAME):
# import the serializers module and track any discovered serializers
serializers_module_name = '%s.%s' % (self.name, SERIALIZERS_MODULE_NAME)
self.serializers_module = import_module(serializers_module_name)
for objname in dir(self.serializers_module):
obj = getattr(self.serializers_module, objname)
try:
# Any subclass of ModelSerializer that isn't itself ModelSerializer
# gets registered in the named_serializers registry.
if (obj is not ModelSerializer and
issubclass(obj, ModelSerializer)):
self.named_serializers[objname] = obj
except TypeError:
# obj isn't a class, issubclass exploded but obj can be safely filtered out
continue
from gettext import gettext as _
from rest_framework import serializers
from pulpcore.app import models
from pulpcore.app.serializers import ModelSerializer
class ProgressReportSerializer(ModelSerializer):
message = serializers.CharField(
help_text=_("The message shown to the user for the progress report."), read_only=True
)
code = serializers.CharField(
help_text=_("Identifies the type of progress report'."), read_only=True
)
state = serializers.CharField(
help_text=_(
"The current state of the progress report. The possible values are:"
" 'waiting', 'skipped', 'running', 'completed', 'failed' and 'canceled'."
" The default is 'waiting'."
),
read_only=True,
)
total = serializers.IntegerField(help_text=_("The total count of items."), read_only=True)
from gettext import gettext as _
from rest_framework import serializers
from pulpcore.app import models
from pulpcore.app.serializers import ModelSerializer, DetailNestedHyperlinkedRelatedField
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"
)
if user_provided_realpath.startswith(allowed_path):
if check_is_dir: # fail if exists and not-directory
if os.path.exists(user_provided_realpath) and not os.path.isdir(
user_provided_realpath
):
raise serializers.ValidationError(
_("Path '{}' must be a directory " "path").format(value)
)
return value
raise serializers.ValidationError(
_("Path '{}' is not an allowed export " "path").format(value)
)
class Meta:
model = models.Exporter
fields = ModelSerializer.Meta.fields + ("name",)
class ExportedResourcesField(serializers.ListField):
def to_representation(self, obj):
result = []
exported_resources = obj.exported_resources.all()
for exported_resource in exported_resources:
viewset = get_viewset_for_model(exported_resource.content_object)
serializer = viewset.serializer_class(
exported_resource.content_object, context={"request": None}
)
result.append(serializer.data.get("pulp_href"))
return result
class ExportSerializer(ModelSerializer):
model = models.Task
fields = ("state",)
class ContentAppStatusSerializer(ModelSerializer):
name = serializers.CharField(help_text=_("The name of the worker."), read_only=True)
last_heartbeat = serializers.DateTimeField(
help_text=_("Timestamp of the last time the worker talked to the service."), read_only=True
)
class Meta:
model = models.ContentAppStatus
fields = ("name", "last_heartbeat")
class WorkerSerializer(ModelSerializer):
pulp_href = IdentityField(view_name="workers-detail")
name = serializers.CharField(help_text=_("The name of the worker."), read_only=True)
last_heartbeat = serializers.DateTimeField(
help_text=_("Timestamp of the last time the worker talked to the service."), read_only=True
)
# disable "created" because we don't care about it
created = None
class Meta:
model = models.Worker
_base_fields = tuple(set(ModelSerializer.Meta.fields) - set(["created"]))
fields = _base_fields + ("name", "last_heartbeat")