Merge pull request #355 from dkliban/handler

Problem: registry code duplicates code from pulpcore

Author: ipanova

pulp_docker/app/downloaders.py

@@ -47,6 +47,7 @@ async def _run(self, handle_401=True, extra_data=None):
             handle_401(bool): If true, catch 401, request a new token and retry.
         """
         headers = {}
+        repo_name = None
         if extra_data is not None:
             headers = extra_data.get('headers', headers)
             repo_name = extra_data.get('repo_name', None)

pulp_docker/app/registry.py

@@ -1,45 +1,19 @@
 import logging
 import os
 
-from aiohttp import web, web_exceptions
-from aiohttp.client_exceptions import ClientResponseError
+from aiohttp import web
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist
-from django.db import IntegrityError, transaction
-from gettext import gettext as _
 from multidict import MultiDict
 
-from pulpcore.plugin.models import Artifact, ContentArtifact, Remote
+from pulpcore.plugin.content import Handler, PathNotResolved
+from pulpcore.plugin.models import ContentArtifact
 from pulp_docker.app.models import DockerDistribution, ManifestTag, ManifestListTag, MEDIA_TYPE
 
 
 log = logging.getLogger(__name__)
 
 
-HOP_BY_HOP_HEADERS = [
-    'connection',
-    'keep-alive',
-    'public',
-    'proxy-authenticate',
-    'transfer-encoding',
-    'upgrade',
-]
-
-
-class PathNotResolved(web_exceptions.HTTPNotFound):
-    """
-    The path could not be resolved to a published file.
-
-    This could be caused by either the distribution, the publication,
-    or the published file could not be found.
-    """
-
-    def __init__(self, path, *args, **kwargs):
-        """Initialize the Exception."""
-        self.path = path
-        super().__init__(*args, **kwargs)
-
-
 class ArtifactNotFound(Exception):
     """
     The artifact associated with a published-artifact does not exist.
@@ -48,16 +22,12 @@ class ArtifactNotFound(Exception):
     pass
 
 
-class Registry:
+class Registry(Handler):
     """
     A set of handlers for the Docker v2 API.
     """
 
-    def __init__(self):
-        """
-        Initializes the Registry class.
-        """
-        self.distribution_model = DockerDistribution
+    distribution_model = DockerDistribution
 
     @staticmethod
     async def get_accepted_media_types(request):
@@ -77,25 +47,19 @@ async def get_accepted_media_types(request):
                 accepted_media_types.append(value.decode('UTF-8'))
         return accepted_media_types
 
-    async def match_distribution(self, path):
+    @staticmethod
+    def _base_paths(path):
         """
-        Match a distribution using a base path.
+        Get a list of base paths used to match a distribution.
 
         Args:
             path (str): The path component of the URL.
 
         Returns:
-            DockerDistribution: The matched docker distribution.
-
-        Raises:
-            PathNotResolved: when not matched.
+            list: Of base paths.
 
         """
-        try:
-            return self.distribution_model.objects.get(base_path=path)
-        except ObjectDoesNotExist:
-            log.debug(_('DockerDistribution not matched for {path}.').format(path=path))
-            raise PathNotResolved(path)
+        return [path]
 
     @staticmethod
     async def _dispatch(path, headers):
@@ -136,7 +100,7 @@ async def tags_list(self, request):
         Handler for Docker Registry v2 tags/list API.
         """
         path = request.match_info['path']
-        distribution = await self.match_distribution(path)
+        distribution = self._match_distribution(path)
         tags = {'name': path, 'tags': set()}
         repository_version = distribution.get_repository_version()
         for c in repository_version.content:
@@ -164,7 +128,7 @@ async def get_tag(self, request):
         """
         path = request.match_info['path']
         tag_name = request.match_info['tag_name']
-        distribution = await self.match_distribution(path)
+        distribution = self._match_distribution(path)
         repository_version = distribution.get_repository_version()
         accepted_media_types = await Registry.get_accepted_media_types(request)
         if MEDIA_TYPE.MANIFEST_LIST in accepted_media_types:
@@ -236,7 +200,7 @@ async def get_by_digest(self, request):
         """
         path = request.match_info['path']
         digest = "sha256:{digest}".format(digest=request.match_info['digest'])
-        distribution = await self.match_distribution(path)
+        distribution = self._match_distribution(path)
         repository_version = distribution.get_repository_version()
         log.info(digest)
         try:
@@ -253,152 +217,3 @@ async def get_by_digest(self, request):
                                                 headers)
             else:
                 return await self._stream_content_artifact(request, web.StreamResponse(), ca)
-
-    async def _stream_content_artifact(self, request, response, content_artifact):
-        """
-        Stream and optionally save a ContentArtifact by requesting it using the associated remote.
-
-        If a fatal download failure occurs while downloading and there are additional
-        :class:`~pulpcore.plugin.models.RemoteArtifact` objects associated with the
-        :class:`~pulpcore.plugin.models.ContentArtifact` they will also be tried. If all
-        :class:`~pulpcore.plugin.models.RemoteArtifact` downloads raise exceptions, an HTTP 502
-        error is returned to the client.
-
-        Args:
-            request(:class:`~aiohttp.web.Request`): The request to prepare a response for.
-            response (:class:`~aiohttp.web.StreamResponse`): The response to stream data to.
-            content_artifact (:class:`~pulpcore.plugin.models.ContentArtifact`): The ContentArtifact
-                to fetch and then stream back to the client
-
-        Raises:
-            :class:`~aiohttp.web.HTTPNotFound` when no
-                :class:`~pulpcore.plugin.models.RemoteArtifact` objects associated with the
-                :class:`~pulpcore.plugin.models.ContentArtifact` returned the binary data needed for
-                the client.
-
-        """
-        for remote_artifact in content_artifact.remoteartifact_set.all():
-            try:
-                response = await self._stream_remote_artifact(request, response, remote_artifact)
-
-            except ClientResponseError:
-                continue
-
-        raise web_exceptions.HTTPNotFound()
-
-    async def _stream_remote_artifact(self, request, response, remote_artifact):
-        """
-        Stream and save a RemoteArtifact.
-
-        Args:
-            request(:class:`~aiohttp.web.Request`): The request to prepare a response for.
-            response (:class:`~aiohttp.web.StreamResponse`): The response to stream data to.
-            content_artifact (:class:`~pulpcore.plugin.models.ContentArtifact`): The ContentArtifact
-                to fetch and then stream back to the client
-
-        Raises:
-            :class:`~aiohttp.web.HTTPNotFound` when no
-                :class:`~pulpcore.plugin.models.RemoteArtifact` objects associated with the
-                :class:`~pulpcore.plugin.models.ContentArtifact` returned the binary data needed for
-                the client.
-
-        """
-        remote = remote_artifact.remote.cast()
-
-        async def handle_headers(headers):
-            for name, value in headers.items():
-                if name.lower() in HOP_BY_HOP_HEADERS:
-                    continue
-                response.headers[name] = value
-            await response.prepare(request)
-
-        async def handle_data(data):
-            await response.write(data)
-            if remote.policy != Remote.STREAMED:
-                await original_handle_data(data)
-
-        async def finalize():
-            if remote.policy != Remote.STREAMED:
-                await original_finalize()
-
-        repo_name = remote.namespaced_upstream_name
-        downloader = remote.get_downloader(remote_artifact=remote_artifact,
-                                           headers_ready_callback=handle_headers)
-        original_handle_data = downloader.handle_data
-        downloader.handle_data = handle_data
-        original_finalize = downloader.finalize
-        downloader.finalize = finalize
-        download_result = await downloader.run(extra_data={'repo_name': repo_name})
-
-        if remote.policy != Remote.STREAMED:
-            self._save_artifact(download_result, remote_artifact)
-        await response.write_eof()
-        return response
-
-    def _save_artifact(self, download_result, remote_artifact):
-        """
-        Create/Get an Artifact and associate it to a RemoteArtifact and/or ContentArtifact.
-
-        Create (or get if already existing) an :class:`~pulpcore.plugin.models.Artifact`
-        based on the `download_result` and associate it to the `content_artifact` of the given
-        `remote_artifact`. Both the created artifact and the updated content_artifact are saved to
-        the DB.  The `remote_artifact` is also saved for the pull-through caching use case.
-
-        Plugin-writers may overide this method if their content module requires
-        additional/different steps for saving.
-
-        Args:
-            download_result (:class:`~pulpcore.plugin.download.DownloadResult`: The
-                DownloadResult for the downloaded artifact.
-
-            remote_artifact (:class:`~pulpcore.plugin.models.RemoteArtifact`): The
-                RemoteArtifact to associate the Artifact with.
-
-        Returns:
-            The associated :class:`~pulpcore.plugin.models.Artifact`.
-
-        """
-        content_artifact = remote_artifact.content_artifact
-        remote = remote_artifact.remote
-        artifact = Artifact(
-            **download_result.artifact_attributes,
-            file=download_result.path
-        )
-        with transaction.atomic():
-            try:
-                with transaction.atomic():
-                    artifact.save()
-            except IntegrityError:
-                artifact = Artifact.objects.get(artifact.q())
-            update_content_artifact = True
-            if content_artifact._state.adding:
-                # This is the first time pull-through content was requested.
-                rel_path = content_artifact.relative_path
-                c_type = remote.get_remote_artifact_content_type(rel_path)
-                content = c_type.init_from_artifact_and_relative_path(artifact, rel_path)
-                try:
-                    with transaction.atomic():
-                        content.save()
-                        content_artifact.content = content
-                        content_artifact.save()
-                except IntegrityError:
-                    # There is already content for this Artifact
-                    content = c_type.objects.get(content.q())
-                    artifacts = content._artifacts
-                    if artifact.sha256 != artifacts[0].sha256:
-                        raise RuntimeError("The Artifact downloaded during pull-through does not "
-                                           "match the Artifact already stored for the same "
-                                           "content.")
-                    content_artifact = ContentArtifact.objects.get(content=content)
-                    update_content_artifact = False
-                try:
-                    with transaction.atomic():
-                        remote_artifact.content_artifact = content_artifact
-                        remote_artifact.save()
-                except IntegrityError:
-                    # Remote artifact must have already gotten saved during a parallel request
-                    log.info("RemoteArtifact already exists.")
-            if update_content_artifact:
-                content_artifact.artifact = artifact
-                content_artifact.save()
-        return artifact