pantsbuild / pants / 19015773527

from __future__ import annotations

import logging

from collections import defaultdict

from dataclasses import dataclass

from itertools import chain

from typing import DefaultDict, cast

from pants.backend.docker.goals.package_image import BuiltDockerImage

from pants.backend.docker.subsystems.docker_options import DockerOptions

from pants.backend.docker.target_types import DockerImageRegistriesField, DockerImageSkipPushField

from pants.backend.docker.util_rules.docker_binary import DockerBinary

from pants.core.goals.publish import (

from pants.core.util_rules.env_vars import environment_vars_subset

from pants.engine.env_vars import EnvironmentVarsRequest

from pants.engine.process import InteractiveProcess, Process

from pants.engine.rules import collect_rules, implicitly, rule

logger = logging.getLogger(__name__)

class PublishDockerImageRequest(PublishRequest):

    pass

@dataclass(frozen=True)

class PublishDockerImageFieldSet(PublishFieldSet):

    publish_request_type = PublishDockerImageRequest

    required_fields = (DockerImageRegistriesField,)

    registries: DockerImageRegistriesField

    skip_push: DockerImageSkipPushField

    def get_output_data(self) -> PublishOutputData:

        return PublishOutputData(

@rule

async def push_docker_images(

    tags = tuple(

    if request.field_set.skip_push.value:

        return PublishProcesses(

    env = await environment_vars_subset(

    skip_push_reasons: DefaultDict[str, DefaultDict[str, set[str]]] = defaultdict(

    jobs: list[PublishPackages] = []

    refs: list[str] = []

    processes: list[Process | InteractiveProcess] = []

    for tag in tags:

        for registry in options.registries().registries.values():

            if registry.skip_push and tag.startswith(f"{registry.address}/"):

                skip_push_reasons["skip_push"][registry.alias].add(tag)

                break

            if registry.use_local_alias and tag.startswith(f"{registry.alias}/"):

                skip_push_reasons["use_local_alias"][registry.alias].add(tag)

                break

            refs.append(tag)

            push_process = docker.push_image(tag, env)

            if options.publish_noninteractively:

                processes.append(push_process)

                processes.append(InteractiveProcess.from_process(push_process))

    for ref, process in zip(refs, processes):

        jobs.append(

    for reason, skip_push in skip_push_reasons.items():

        for name, skip_tags in skip_push.items():

            jobs.append(

    return PublishProcesses(jobs)

def rules():

    return (