feat: new CLI: airbyte-cdk image build (#504)

Co-authored-by: octavia-squidington-iii <[email protected]>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: 3 people

airbyte_cdk/cli/__init__.py

@@ -1 +1,9 @@
-# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""The `airbyte-cdk.cli` module provides command-line interfaces for the Airbyte CDK.
+
+As of now, it includes the following CLI entry points:
+
+- `airbyte-cdk`: Commands for working with connectors.
+- `source-declarative-manifest`: Directly invoke the declarative manifests connector.
+
+"""

airbyte_cdk/cli/airbyte_cdk/_image.py

@@ -1,22 +1,93 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
-"""Docker image commands.
+"""Airbyte CDK 'image' commands.
 
-Coming soon.
-
-This module is planned to provide a command line interface (CLI) for building
-Docker images for Airbyte CDK connectors.
+The `airbyte-cdk image build` command provides a simple way to work with Airbyte
+connector images.
 """
 
-import click
+import sys
+from pathlib import Path
+
+import rich_click as click
+
+from airbyte_cdk.cli.airbyte_cdk._util import resolve_connector_name_and_directory
+from airbyte_cdk.models.connector_metadata import MetadataFile
+from airbyte_cdk.utils.docker import (
+    ConnectorImageBuildError,
+    build_connector_image,
+    verify_docker_installation,
+)
 
 
 @click.group(
     name="image",
     help=__doc__.replace("\n", "\n\n"),  # Render docstring as help text (markdown)
 )
 def image_cli_group() -> None:
-    """Docker image commands."""
-    pass
+    """Commands for working with connector Docker images."""
+
+
+@image_cli_group.command()
+@click.option(
+    "--connector-name",
+    type=str,
+    help="Name of the connector to test. Ignored if --connector-directory is provided.",
+)
+@click.option(
+    "--connector-directory",
+    type=click.Path(exists=True, file_okay=False, path_type=Path),
+    help="Path to the connector directory.",
+)
+@click.option("--tag", default="dev", help="Tag to apply to the built image (default: dev)")
+@click.option("--no-verify", is_flag=True, help="Skip verification of the built image")
+def build(
+    connector_name: str | None = None,
+    connector_directory: Path | None = None,
+    *,
+    tag: str = "dev",
+    no_verify: bool = False,
+) -> None:
+    """Build a connector Docker image.
+
+    This command builds a Docker image for a connector, using either
+    the connector's Dockerfile or a base image specified in the metadata.
+    The image is built for both AMD64 and ARM64 architectures.
+    """
+    if not verify_docker_installation():
+        click.echo(
+            "Docker is not installed or not running. Please install Docker and try again.", err=True
+        )
+        sys.exit(1)
+
+    connector_name, connector_directory = resolve_connector_name_and_directory(
+        connector_name=connector_name,
+        connector_directory=connector_directory,
+    )
+
+    metadata_file_path: Path = connector_directory / "metadata.yaml"
+    try:
+        metadata = MetadataFile.from_file(metadata_file_path)
+    except (FileNotFoundError, ValueError) as e:
+        click.echo(
+            f"Error loading metadata file '{metadata_file_path}': {e!s}",
+            err=True,
+        )
+        sys.exit(1)
+    click.echo(f"Building Image for Connector: {metadata.data.dockerRepository}:{tag}")
+    try:
+        build_connector_image(
+            connector_directory=connector_directory,
+            connector_name=connector_name,
+            metadata=metadata,
+            tag=tag,
+            no_verify=no_verify,
+        )
+    except ConnectorImageBuildError as e:
+        click.echo(
+            f"Error building connector image: {e!s}",
+            err=True,
+        )
+        sys.exit(1)
 
 
 __all__ = [

airbyte_cdk/models/connector_metadata.py

@@ -0,0 +1,97 @@
+"""Models to represent the structure of a `metadata.yaml` file."""
+
+from __future__ import annotations
+
+from enum import Enum
+from pathlib import Path
+
+import yaml
+from pydantic import BaseModel, Field
+
+
+class ConnectorLanguage(str, Enum):
+    """Connector implementation language."""
+
+    PYTHON = "python"
+    JAVA = "java"
+    LOW_CODE = "low-code"
+    MANIFEST_ONLY = "manifest-only"
+    UNKNOWN = "unknown"
+
+
+class ConnectorBuildOptions(BaseModel):
+    """Connector build options from metadata.yaml."""
+
+    model_config = {"extra": "allow"}
+
+    baseImage: str | None = Field(
+        None,
+        description="Base image to use for building the connector",
+    )
+    path: str | None = Field(
+        None,
+        description="Path to the connector code within the repository",
+    )
+
+
+class ConnectorMetadata(BaseModel):
+    """Connector metadata from metadata.yaml."""
+
+    model_config = {"extra": "allow"}
+
+    dockerRepository: str = Field(..., description="Docker repository for the connector image")
+    dockerImageTag: str = Field(..., description="Docker image tag for the connector")
+
+    tags: list[str] = Field(
+        default=[],
+        description="List of tags for the connector",
+    )
+
+    @property
+    def language(self) -> ConnectorLanguage:
+        """Get the connector language."""
+        for tag in self.tags:
+            if tag.startswith("language:"):
+                language = tag.split(":", 1)[1]
+                if language == "python":
+                    return ConnectorLanguage.PYTHON
+                elif language == "java":
+                    return ConnectorLanguage.JAVA
+                elif language == "low-code":
+                    return ConnectorLanguage.LOW_CODE
+                elif language == "manifest-only":
+                    return ConnectorLanguage.MANIFEST_ONLY
+
+        return ConnectorLanguage.UNKNOWN
+
+    connectorBuildOptions: ConnectorBuildOptions | None = Field(
+        None, description="Options for building the connector"
+    )
+
+
+class MetadataFile(BaseModel):
+    """Represents the structure of a metadata.yaml file."""
+
+    model_config = {"extra": "allow"}
+
+    data: ConnectorMetadata = Field(..., description="Connector metadata")
+
+    @classmethod
+    def from_file(
+        cls,
+        file_path: Path,
+    ) -> MetadataFile:
+        """Load metadata from a YAML file."""
+        if not file_path.exists():
+            raise FileNotFoundError(f"Metadata file not found: {file_path!s}")
+
+        metadata_content = file_path.read_text()
+        metadata_dict = yaml.safe_load(metadata_content)
+
+        if not metadata_dict or "data" not in metadata_dict:
+            raise ValueError(
+                "Invalid metadata format: missing 'data' field in YAML file '{file_path!s}'"
+            )
+
+        metadata_file = MetadataFile.model_validate(metadata_dict)
+        return metadata_file