Method ensure

misc/python/materialize/mzbuild.py:1426–1484

Ensure all publishable images in this dependency set exist on Docker Hub. Images are pushed using their spec as their tag. Args: pre_build: A callback to invoke with all dependency that are going to be built locally, invoked after their ca

(self, pre_build: Callable[[list[ResolvedImage]], None] | None = None)

            dep.build(prep)

    def ensure(self, pre_build: Callable[[list[ResolvedImage]], None] | None = None):
        """Ensure all publishable images in this dependency set exist on Docker
        Hub.

        Images are pushed using their spec as their tag.

        Args:
            pre_build: A callback to invoke with all dependency that are going
                       to be built locally, invoked after their cargo build is
                       done, but before the Docker images are build and
                       uploaded to DockerHub.
        """
        num_deps = len(list(self))
        if not num_deps:
            deps_to_build = []
        else:
            with ThreadPoolExecutor(max_workers=num_deps) as executor:
                futures = list(
                    executor.map(
                        lambda dep: (dep, not dep.is_published_if_necessary()), self
                    )
                )

            deps_to_build = [dep for dep, should_build in futures if should_build]

        prep = self._prepare_batch(deps_to_build)
        if pre_build:
            pre_build(deps_to_build)
        lock = Lock()
        built_deps: set[str] = set([dep.name for dep in self]) - set(
            [dep.name for dep in deps_to_build]
        )

        def build_dep(dep):
            end_time = time.time() + 600
            while True:
                if time.time() > end_time:
                    raise TimeoutError(
                        f"Timed out in {dep.name} waiting for {[dep2 for dep2 in dep.dependencies if dep2 not in built_deps]}"
                    )
                with lock:
                    if all(dep2 in built_deps for dep2 in dep.dependencies):
                        break
                time.sleep(0.01)
            for attempts_remaining in reversed(range(3)):
                try:
                    dep.build(prep, push=dep.publish)
                    with lock:
                        built_deps.add(dep.name)
                    break
                except Exception:
                    if not dep.publish or attempts_remaining == 0:
                        raise

        if deps_to_build:
            with ThreadPoolExecutor(max_workers=len(deps_to_build)) as executor:
                futures = [executor.submit(build_dep, dep) for dep in deps_to_build]
                for future in as_completed(futures):