Story #8068 » dynamicload.patch
| pulp_container/app/registry_api.py | ||
|---|---|---|
|
import logging
|
||
|
import hashlib
|
||
|
import re
|
||
|
from collections import namedtuple
|
||
|
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
|
||
|
from tempfile import NamedTemporaryFile
|
||
| ... | ... | |
|
from rest_framework.views import APIView
|
||
|
from pulp_container.app import models, serializers
|
||
|
from pulp_container.app.access_policy import RegistryAccessPolicy
|
||
|
from pulp_container.app.authorization import AuthorizationService
|
||
|
from pulp_container.app.redirects import FileStorageRedirects, S3StorageRedirects
|
||
|
from pulp_container.app.token_verification import (
|
||
| ... | ... | |
|
TokenPermission,
|
||
|
)
|
||
|
FakeView = namedtuple("FakeView", ["action", "get_object"])
|
||
|
log = logging.getLogger(__name__)
|
||
| ... | ... | |
|
Handles requests to the /v2/_catalog endpoint
|
||
|
"""
|
||
|
queryset = models.ContainerDistribution.objects.all().only("base_path")
|
||
|
queryset = models.ContainerDistribution.objects.all().only("base_path", "pk")
|
||
|
serializer_class = ContainerCatalogSerializer
|
||
|
pagination_class = ContainerCatalogPagination
|
||
|
permission_class = RegistryAccessPolicy()
|
||
|
def get_queryset(self, *args, **kwargs):
|
||
|
queryset = super().get_queryset()
|
||
|
accessible_pks = []
|
||
|
for obj in queryset:
|
||
|
view = FakeView("pull", lambda: obj)
|
||
|
if self.permission_class.has_permission(self.request, view):
|
||
|
accessible_pks.append(obj.pk)
|
||
|
return queryset.filter(pk__in=accessible_pks)
|
||
|
class ContainerTagListSerializer(ModelSerializer):
|
||
.