async def get(self, provider_prefix, _unescaped_spec):
        """Get a built image for a given spec and repo provider.

        Different repo providers will require different spec information. This
        function relies on the functionality of the tornado `GET` request.

        Parameters
        ----------
            provider_prefix : str
                the nickname for a repo provider (i.e. 'gh')
            spec:
                specifies information needed by the repo provider (i.e. user,
                repo, ref, etc.)

        """
        prefix = "/build/" + provider_prefix
        spec = self.get_spec_from_request(prefix)

        # verify the build token and rate limit
        build_token = self.get_argument("build_token", None)
        self.check_build_token(build_token, f"{provider_prefix}/{spec}")
        self.check_rate_limit()

        # Verify if the provider is valid for EventSource.
        # EventSource cannot handle HTTP errors, so we must validate and send
        # error messages on the eventsource.
        if provider_prefix not in self.settings["repo_providers"]:
            await self.fail(f"No provider found for prefix {provider_prefix}")
            return

        # create a heartbeat
        IOLoop.current().spawn_callback(self.keep_alive)

        spec = spec.rstrip("/")
        key = f"{provider_prefix}:{spec}"

        # get a provider object that encapsulates the provider and the spec
        try:
            provider = self.get_provider(provider_prefix, spec=spec)
        except Exception as e:
            app_log.exception("Failed to get provider for %s", key)
            await self.fail(str(e))
            return

        if provider.is_banned():
            await self.emit(
                {
                    "phase": "failed",
                    "message": f"Sorry, {spec} is not allowed to launch. Please contact admins for more info!",
                }
            )
            return

        repo_url = self.repo_url = provider.get_repo_url()

        # labels to apply to build/launch metrics
        self.repo_metric_labels = {
            "provider": provider.name,
            "repo": repo_url,
        }

        try:
            ref = await provider.get_resolved_ref()
        except Exception as e:
            await self.fail(f"Error resolving ref for {key}: {e}")
            return

        if ref is None:
            error_message = [
                f"Could not resolve ref for {key}. Double check your URL and that your repo is public."
            ]

            if provider.name == "GitHub":
                if provider.unresolved_ref in {"master", "main"}:
                    if provider.unresolved_ref == "master":
                        error_message.append(
                            'GitHub has changed default branches from "master" to "main".'
                        )
                    error_message.append(
                        "Tip: HEAD will always resolve to a repository's default branch."
                    )

                    # keep old links working for default branch names
                    # by substituting 'master' or 'main' with 'HEAD'
                    pre_ref_spec, _ = spec.rsplit("/", 1)
                    spec = f"{pre_ref_spec}/HEAD"
                    unresolved_ref = provider.unresolved_ref
                    try:
                        provider = self.get_provider(provider_prefix, spec=spec)
                        ref = await provider.get_resolved_ref()
                    except Exception as e:
                        # if this fails, leave ref as None, which will fail below
                        self.log.error(f"Error redirecting {key} to HEAD: {e}")
                    else:
                        # delayed redirect for deleted default branches
                        await self.emit(
                            {
                                "phase": "waiting",
                                "message": (
                                    " ".join(error_message) + "\n"
                                    f"Trying again with HEAD instead of {unresolved_ref}. Please update your links.\n"
                                ),
                            }
                        )
                        # artificial delay for what should be broken links
                        await asyncio.sleep(10)

            if ref is None:
                # ref can become non-None if redirected to HEAD
                await self.fail(" ".join(error_message))
                return

        self.ref_url = await provider.get_resolved_ref_url()
        resolved_spec = await provider.get_resolved_spec()

        badge_base_url = self.get_badge_base_url()
        self.binder_launch_host = badge_base_url or "{proto}://{host}{base_url}".format(
            proto=self.request.protocol,
            host=self.request.host,
            base_url=self.settings["base_url"],
        )
        # These are relative URLs so do not have a leading /
        self.binder_request = "v2/{provider}/{spec}".format(
            provider=provider_prefix,
            spec=spec,
        )
        self.binder_persistent_request = "v2/{provider}/{spec}".format(
            provider=provider_prefix,
            spec=resolved_spec,
        )

        # generate a complete build name (for GitHub: `build-{user}-{repo}-{ref}`)

        image_prefix = self.settings["image_prefix"]

        # Enforces max 255 characters before image
        safe_build_slug = _safe_build_slug(
            provider.get_build_slug(), limit=255 - len(image_prefix)
        )

        build_name = _generate_build_name(
            provider.get_build_slug(), ref, prefix="build-"
        )

        image_name = self.image_name = (
            "{prefix}{build_slug}:{ref}".format(
                prefix=image_prefix, build_slug=safe_build_slug, ref=ref
            )
            .replace("_", "-")
            .lower()
        )

        image_without_tag, image_tag = _get_image_basename_and_tag(image_name)
        if self.settings["use_registry"]:
            for _ in range(3):
                try:
                    image_manifest = await self.registry.get_image_manifest(
                        image_without_tag, image_tag
                    )
                    image_found = bool(image_manifest)
                    break
                except HTTPClientError:
                    app_log.exception(
                        "Failed to get image manifest for %s",
                        image_name,
                    )
                    image_found = False
        else:
            # Check if the image exists locally!
            # Assume we're running in single-node mode or all binder pods are assigned to the same node!
            docker_client = docker.from_env(version="auto")
            try:
                docker_client.images.get(image_name)
            except docker.errors.ImageNotFound:
                # image doesn't exist, so do a build!
                image_found = False
            else:
                image_found = True

        build_only = self._get_build_only()
        if image_found:
            if build_only:
                await self.emit(
                    {
                        "phase": "ready",
                        "imageName": image_name,
                        "message": "Done! Found built image\n",
                    }
                )
            else:
                await self.emit(
                    {
                        "phase": "built",
                        "imageName": image_name,
                        "message": "Found built image, launching...\n",
                    }
                )
                with LAUNCHES_INPROGRESS.track_inprogress():
                    try:
                        await self.launch(provider)
                    except LaunchQuotaExceeded:
                        return
                self.event_log.emit(
                    "binderhub.jupyter.org/launch",
                    5,
                    {
                        "provider": provider.name,
                        "spec": spec,
                        "ref": ref,
                        "status": "success",
                        "build_token": self._have_build_token,
                        "origin": (
                            self.settings["normalized_origin"]
                            if self.settings["normalized_origin"]
                            else self.request.host
                        ),
                    },
                )
            return

        # Don't allow builds when quota is exceeded
        try:
            await self.check_quota(provider)
        except LaunchQuotaExceeded:
            return

        # Prepare to build
        q = Queue()

        BuildClass = self.settings.get("build_class")

        build = BuildClass(
            # All other properties should be set in traitlets config
            parent=self.settings["traitlets_parent"],
            q=q,
            name=build_name,
            repo_url=repo_url,
            ref=ref,
            image_name=image_name,
            git_credentials=provider.git_credentials,
        )
        if self.settings["use_registry"]:
            push_token = await self.registry.get_credentials(
                image_without_tag, image_tag
            )
            if push_token:
                build.registry_credentials = push_token
        else:
            build.push_secret = ""

        self.build = build

        with BUILDS_INPROGRESS.track_inprogress():
            done = False
            failed = False

            def _check_result(future):
                nonlocal done
                nonlocal failed
                try:
                    r = future.result()
                    app_log.debug("task completed: %s", r)
                except Exception:
                    app_log.error("task failed", exc_info=True)
                    done = True
                    failed = True
                    # TODO: Propagate error to front-end

            build_starttime = time.perf_counter()
            pool = self.settings["build_pool"]
            # Start building
            submit_future = pool.submit(build.submit)
            submit_future.add_done_callback(_check_result)
            IOLoop.current().add_callback(lambda: submit_future)

            log_future = None

            # initial waiting event
            await self.emit(
                {
                    "phase": "waiting",
                    "message": "Waiting for build to start...\n",
                }
            )

            while not done:
                progress = await q.get()
                # FIXME: If pod goes into an unrecoverable stage, such as ImagePullBackoff or
                # whatever, we should fail properly.
                if progress.kind == ProgressEvent.Kind.BUILD_STATUS_CHANGE:
                    phase = progress.payload.value
                    if progress.payload == ProgressEvent.BuildStatus.PENDING:
                        # nothing to do, just waiting
                        continue
                    elif progress.payload == ProgressEvent.BuildStatus.BUILT:
                        if build_only:
                            message = "Done! Image built\n"
                            phase = "ready"
                        else:
                            message = "Built image, launching...\n"
                        event = {
                            "phase": phase,
                            "message": message,
                            "imageName": image_name,
                        }
                        BUILD_TIME.labels(status="success").observe(
                            time.perf_counter() - build_starttime
                        )
                        BUILD_COUNT.labels(
                            status="success", **self.repo_metric_labels
                        ).inc()
                        done = True
                    elif progress.payload == ProgressEvent.BuildStatus.RUNNING:
                        # start capturing build logs once the pod is running
                        if log_future is None:
                            log_future = pool.submit(build.stream_logs)
                            log_future.add_done_callback(_check_result)
                        continue
                    elif progress.payload == ProgressEvent.BuildStatus.BUILT:
                        # Do nothing, is ok!
                        continue
                    elif progress.payload == ProgressEvent.BuildStatus.FAILED:
                        event = {"phase": phase}
                    elif progress.payload == ProgressEvent.BuildStatus.UNKNOWN:
                        event = {"phase": phase}
                    else:
                        raise ValueError(
                            f"Found unknown phase {phase} in ProgressEvent"
                        )
                elif progress.kind == ProgressEvent.Kind.LOG_MESSAGE:
                    # The logs are coming out of repo2docker, so we expect
                    # them to be JSON structured anyway
                    event = progress.payload
                    payload = json.loads(event)
                    if payload.get("phase") in ("failure", "failed"):
                        failed = True
                        BUILD_TIME.labels(status="failure").observe(
                            time.perf_counter() - build_starttime
                        )
                        BUILD_COUNT.labels(
                            status="failure", **self.repo_metric_labels
                        ).inc()
                await self.emit(event)

        if build_only:
            return

        if not failed:
            # Launch after building an image
            with LAUNCHES_INPROGRESS.track_inprogress():
                await self.launch(provider)
            self.event_log.emit(
                "binderhub.jupyter.org/launch",
                5,
                {
                    "provider": provider.name,
                    "spec": spec,
                    "ref": ref,
                    "status": "success",
                    "build_token": self._have_build_token,
                    "origin": (
                        self.settings["normalized_origin"]
                        if self.settings["normalized_origin"]
                        else self.request.host
                    ),
                },
            )

        # Don't close the eventstream immediately.
        # (javascript) eventstream clients reconnect automatically on dropped connections,
        # so if the server closes the connection first,
        # the client will reconnect which starts a new build.
        # If we sleep here, that makes it more likely that a well-behaved
        # client will close its connection first.
        # The duration of this shouldn't matter because
        # well-behaved clients will close connections after they receive the launch event.
        await gen.sleep(60)
Get a built image for a given spec and repo provider.

        Different repo providers will require different spec information. This
        function relies on the functionality of the tornado `GET` request.

        Parameters
        ----------
            provider_prefix : str
                the nickname for a repo provider (i.e. 'gh')
            spec:
                specifies information needed by the repo provider (i.e. user,
                repo, ref, etc.)

        
binderhub/builder.py