Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
- define a `Meta` class which should:
- specify a plugin exporter model for which filter is defined
- extend `fields` with specific ones
"""
name = filters.CharFilter()
last_export = IsoDateTimeFilter()
class Meta:
model = Exporter
fields = {
'name': NAME_FILTER_OPTIONS,
'last_export': DATETIME_FILTER_OPTIONS
}
class ExporterViewSet(NamedModelViewSet,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
AsyncUpdateMixin,
AsyncRemoveMixin):
endpoint_name = 'exporters'
serializer_class = ExporterSerializer
queryset = Exporter.objects.all()
filterset_class = ExporterFilter
mixins.UpdateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
mixins.DestroyModelMixin,
):
"""ViewSet for Importers."""
queryset = Importer.objects.all()
serializer_class = ImporterSerializer
endpoint_name = "importers"
router_lookup = "importer"
filterset_class = ImporterFilter
class ImportViewSet(
NamedModelViewSet,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
mixins.DestroyModelMixin,
):
"""ViewSet for viewing imports from an Importer."""
endpoint_name = "imports"
nest_prefix = "importers"
router_lookup = "import"
lookup_field = "pk"
parent_viewset = ImporterViewSet
parent_lookup_kwargs = {"importer_pk": "importer__pk"}
serializer_class = ImportSerializer
queryset = Import.objects.all()
class ContentFilter(BaseFilterSet):
"""
Plugin content filters should:
- inherit from this class
- add any plugin-specific filters if needed
- define its own `Meta` class which should:
- specify plugin content model
- extend `fields` with plugin-specific ones
"""
class Meta:
model = Content
fields = {'type': ['exact', 'in']}
class ContentViewSet(NamedModelViewSet,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin):
endpoint_name = 'content'
queryset = Content.objects.all()
serializer_class = ContentSerializer
filterset_class = ContentFilter
@transaction.atomic
def create(self, request):
"""
Create a Content Artifact
"""
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
artifacts = serializer.validated_data.pop('artifacts')
def import_viewsets(self):
# circular import avoidance
from pulpcore.app.viewsets import NamedModelViewSet
self.named_viewsets = {}
if module_has_submodule(self.module, VIEWSETS_MODULE_NAME):
# import the viewsets module and track any interesting viewsets
viewsets_module_name = '{name}.{module}'.format(
name=self.name, module=VIEWSETS_MODULE_NAME)
self.viewsets_module = import_module(viewsets_module_name)
for objname in dir(self.viewsets_module):
obj = getattr(self.viewsets_module, objname)
try:
# Any subclass of NamedModelViewSet that isn't itself NamedModelViewSet
# gets registered in the named_viewsets registry.
if (obj is not NamedModelViewSet and issubclass(obj, NamedModelViewSet)):
model = obj.queryset.model
self.named_viewsets[model] = obj
except TypeError:
# obj isn't a class, issubclass exploded but obj can be safely filtered out
continue
- define a `Meta` class which should:
- specify a plugin publisher model for which filter is defined
- extend `fields` with specific ones
"""
name = filters.CharFilter()
_last_updated = IsoDateTimeFilter()
class Meta:
model = Publisher
fields = {
'name': NAME_FILTER_OPTIONS,
'_last_updated': DATETIME_FILTER_OPTIONS
}
class PublisherViewSet(NamedModelViewSet,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
AsyncUpdateMixin,
AsyncRemoveMixin):
endpoint_name = 'publishers'
serializer_class = PublisherSerializer
queryset = Publisher.objects.all()
filterset_class = PublisherFilter
class ExporterFilter(BaseFilterSet):
"""
Plugin exporter filter should:
- inherit from this class
- add any specific filters if needed
if value:
return queryset.filter(pk__in=online_workers)
else:
return queryset.exclude(pk__in=online_workers)
def filter_missing(self, queryset, name, value):
missing_workers = Worker.objects.missing_workers()
if value:
return queryset.filter(pk__in=missing_workers)
else:
return queryset.exclude(pk__in=missing_workers)
class WorkerViewSet(NamedModelViewSet, mixins.RetrieveModelMixin, mixins.ListModelMixin):
queryset = Worker.objects.all()
serializer_class = WorkerSerializer
endpoint_name = "workers"
http_method_names = ["get", "options"]
lookup_value_regex = "[^/]+"
filterset_class = WorkerFilter
finished_at = IsoDateTimeFilter(field_name='finished_at')
parent = HyperlinkRelatedFilter()
class Meta:
model = Task
fields = {
'state': ['exact', 'in'],
'worker': ['exact', 'in'],
'name': ['contains'],
'started_at': DATETIME_FILTER_OPTIONS,
'finished_at': DATETIME_FILTER_OPTIONS,
'parent': ['exact']
}
class TaskViewSet(NamedModelViewSet,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
mixins.DestroyModelMixin):
queryset = Task.objects.all()
endpoint_name = 'tasks'
filterset_class = TaskFilter
serializer_class = TaskSerializer
minimal_serializer_class = MinimalTaskSerializer
filter_backends = (OrderingFilter, DjangoFilterBackend)
ordering = ('-_created')
@detail_route(methods=('post',))
def cancel(self, request, pk=None):
task = self.get_object()
cancel_task(task.pk)
return Response(status=status.HTTP_204_NO_CONTENT)
fields = {
"name": ["contains"],
"state": ["exact", "in"],
"worker": ["exact", "in"],
"started_at": DATETIME_FILTER_OPTIONS,
"finished_at": DATETIME_FILTER_OPTIONS,
"parent_task": ["exact"],
"child_tasks": ["exact"],
"task_group": ["exact"],
"reserved_resources_record": ["exact"],
"created_resources": ["exact"],
}
class TaskViewSet(
NamedModelViewSet, mixins.RetrieveModelMixin, mixins.ListModelMixin, mixins.DestroyModelMixin
):
queryset = Task.objects.all()
endpoint_name = "tasks"
filterset_class = TaskFilter
serializer_class = TaskSerializer
minimal_serializer_class = MinimalTaskSerializer
filter_backends = (OrderingFilter, DjangoFilterBackend)
ordering = "-pulp_created"
@extend_schema(
description="This operation cancels a task.",
summary="Cancel a task",
operation_id="tasks_cancel",
responses={200: TaskSerializer, 409: TaskSerializer},
)
def partial_update(self, request, pk=None, partial=True):
def filter(self, qs, value):
"""
Args:
qs (django.db.models.query.QuerySet): The QuerySet to filter
value (string): The content href to filter by
Returns:
Queryset of the content contained within the specified created resource
"""
if value is None:
return qs
match = resolve(value)
resource = NamedModelViewSet.get_resource(value, match.func.cls.queryset.model)
return qs.filter(created_resources__object_id=resource.pk)
- define a `Meta` class which should:
- specify a plugin remote model for which filter is defined
- extend `fields` with specific ones
"""
name = filters.CharFilter()
_last_updated = IsoDateTimeFilter()
class Meta:
model = Remote
fields = {
'name': NAME_FILTER_OPTIONS,
'_last_updated': DATETIME_FILTER_OPTIONS
}
class RemoteViewSet(NamedModelViewSet,
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
AsyncUpdateMixin,
AsyncRemoveMixin):
endpoint_name = 'remotes'
serializer_class = RemoteSerializer
queryset = Remote.objects.all()
filterset_class = RemoteFilter
class PublisherFilter(BaseFilterSet):
"""
Plugin publisher filter should:
- inherit from this class
- add any specific filters if needed