review comments

.github/actions/run-dagger-pipeline/action.yml

@@ -83,6 +83,9 @@ inputs:
     description: "URL to airbyte-ci binary"
     required: false
     default: https://connectors.airbyte.com/airbyte-ci/releases/ubuntu/latest/airbyte-ci
+  pypi_token:
+    description: "PyPI API token to publish to PyPI"
+    required: false
 
 runs:
   using: "composite"
@@ -182,3 +185,4 @@ runs:
         CI: "True"
         TAILSCALE_AUTH_KEY: ${{ inputs.tailscale_auth_key }}
         DOCKER_REGISTRY_MIRROR_URL: ${{ inputs.docker_registry_mirror_url }}
+        PYPI_TOKEN: ${{ inputs.pypi_token }}

.github/workflows/publish_connectors.yml

@@ -84,8 +84,7 @@ jobs:
           s3_build_cache_secret_key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
           tailscale_auth_key: ${{ secrets.TAILSCALE_AUTH_KEY }}
           subcommand: "connectors ${{ github.event.inputs.connectors-options }} publish ${{ github.event.inputs.publish-options }}"
-        env:
-          PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
+          pypi_token: ${{ secrets.PYPI_TOKEN }}
 
   set-instatus-incident-on-failure:
     name: Create Instatus Incident on Failure

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/publish/pipeline.py

@@ -14,7 +14,7 @@
 from pipelines.airbyte_ci.connectors.publish.context import PublishConnectorContext
 from pipelines.airbyte_ci.connectors.reports import ConnectorReport
 from pipelines.airbyte_ci.metadata.pipeline import MetadataUpload, MetadataValidation
-from pipelines.airbyte_ci.poetry.pipeline import PublishToPyPI, PyPIPublishContext
+from pipelines.airbyte_ci.poetry.publish.pipeline import PublishToPyPI, PyPIPublishContext
 from pipelines.airbyte_ci.poetry.utils import is_package_published
 from pipelines.dagger.actions.remote_storage import upload_to_gcs
 from pipelines.dagger.actions.system import docker
@@ -54,12 +54,12 @@ async def _run(self) -> StepResult:
             return StepResult(self, status=StepStatus.SUCCESS, stdout=f"No manifest found for {self.context.docker_image}.")
 
 
-class CheckPypiPackageExists(Step):
+class CheckPypiPackageDoesNotExist(Step):
     context: PyPIPublishContext
     title = "Check if the connector is published on pypi"
 
     async def _run(self) -> StepResult:
-        is_published = is_package_published(self.context.package_name, self.context.version, self.context.test_pypi)
+        is_published = is_package_published(self.context.package_name, self.context.version, self.context.registry)
         if is_published:
             return StepResult(
                 self, status=StepStatus.SKIPPED, stderr=f"{self.context.package_name} already exists in version {self.context.version}."
@@ -291,21 +291,10 @@ def create_connector_report(results: List[StepResult]) -> ConnectorReport:
                 metadata_upload_results = await metadata_upload_step.run()
                 results.append(metadata_upload_results)
 
-            # Try to convert the context to a PyPIPublishContext. If it returns None, it means we don't need to publish to pypi.
-            pypi_context = await PyPIPublishContext.from_connector_context(context)
-            if pypi_context:
-                check_pypi_package_exists_results = await CheckPypiPackageExists(pypi_context).run()
-                results.append(check_pypi_package_exists_results)
-                if check_pypi_package_exists_results.status is StepStatus.SKIPPED:
-                    context.logger.info("The connector version is already published on pypi.")
-                elif check_pypi_package_exists_results.status is StepStatus.SUCCESS:
-                    context.logger.info("The connector version is not published on pypi. Let's build and publish it.")
-                    publish_to_pypi_results = await PublishToPyPI(pypi_context).run()
-                    results.append(publish_to_pypi_results)
-                    if publish_to_pypi_results.status is StepStatus.FAILURE:
-                        return create_connector_report(results)
-                elif check_pypi_package_exists_results.status is StepStatus.FAILURE:
-                    return create_connector_report(results)
+            pypi_steps, terminate_early = await _run_pypi_publish_pipeline(context)
+            results.extend(pypi_steps)
+            if terminate_early:
+                return create_connector_report(results)
 
             # Exit early if the connector image already exists or has failed to build
             if check_connector_image_results.status is not StepStatus.SUCCESS:
@@ -346,6 +335,33 @@ def create_connector_report(results: List[StepResult]) -> ConnectorReport:
     return connector_report
 
 
+async def _run_pypi_publish_pipeline(context: PublishConnectorContext) -> Tuple[List[StepResult], bool]:
+    """
+    Run the pypi publish pipeline for a single connector.
+    Return the results of the steps and a boolean indicating whether there was an error and the pipeline should be stopped.
+    """
+    results = []
+    # Try to convert the context to a PyPIPublishContext. If it returns None, it means we don't need to publish to pypi.
+    pypi_context = await PyPIPublishContext.from_publish_connector_context(context)
+    if not pypi_context:
+        return results, False
+
+    check_pypi_package_exists_results = await CheckPypiPackageDoesNotExist(pypi_context).run()
+    results.append(check_pypi_package_exists_results)
+    if check_pypi_package_exists_results.status is StepStatus.SKIPPED:
+        context.logger.info("The connector version is already published on pypi.")
+    elif check_pypi_package_exists_results.status is StepStatus.SUCCESS:
+        context.logger.info("The connector version is not published on pypi. Let's build and publish it.")
+        publish_to_pypi_results = await PublishToPyPI(pypi_context).run()
+        results.append(publish_to_pypi_results)
+        if publish_to_pypi_results.status is StepStatus.FAILURE:
+            return results, True
+    elif check_pypi_package_exists_results.status is StepStatus.FAILURE:
+        return results, True
+
+    return results, False
+
+
 def reorder_contexts(contexts: List[PublishConnectorContext]) -> List[PublishConnectorContext]:
     """Reorder contexts so that the ones that are for strict-encrypt/secure connectors come first.
     The metadata upload on publish checks if the the connectors referenced in the metadata file are already published to DockerHub.

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/poetry/commands.py

@@ -16,87 +16,28 @@
 from pipelines.airbyte_ci.format.format_command import FormatCommand
 from pipelines.cli.click_decorators import click_ignore_unused_kwargs, click_merge_args_into_context_obj
 from pipelines.cli.dagger_pipeline_command import DaggerPipelineCommand
+from pipelines.cli.lazy_group import LazyGroup
 from pipelines.helpers.cli import LogOptions, invoke_commands_concurrently, invoke_commands_sequentially, log_command_results
 from pipelines.models.contexts.click_pipeline_context import ClickPipelineContext, pass_pipeline_context
 from pipelines.models.steps import StepStatus
 
-from .pipeline import PublishToPyPI, PyPIPublishContext
-
 
 @click.group(
     name="poetry",
     help="Commands related to running poetry commands.",
+    cls=LazyGroup,
+    lazy_subcommands={
+        "publish": "pipelines.airbyte_ci.poetry.publish.commands.publish",
+    },
 )
 @click.option(
     "--package-path",
     help="The path to publish",
     type=click.STRING,
     required=True,
 )
-@click.option("--docker-image", help="The docker image to run the command in.", type=click.STRING, default="mwalbeck/python-poetry")
 @click_merge_args_into_context_obj
 @pass_pipeline_context
 @click_ignore_unused_kwargs
 async def poetry(pipeline_context: ClickPipelineContext) -> None:
     pass
-
-
-@poetry.command(cls=DaggerPipelineCommand, name="publish", help="Publish a Python package to PyPI.")
-@click.option(
-    "--pypi-token",
-    help="Access token",
-    type=click.STRING,
-    required=True,
-    envvar="PYPI_TOKEN",
-)
-@click.option(
-    "--test-pypi",
-    help="Whether to publish to test.pypi.org instead of pypi.org.",
-    type=click.BOOL,
-    is_flag=True,
-    default=False,
-)
-@click.option(
-    "--publish-name",
-    help="The name of the package to publish. If not set, the name will be inferred from the pyproject.toml file of the package.",
-    type=click.STRING,
-)
-@click.option(
-    "--publish-version",
-    help="The version of the package to publish. If not set, the version will be inferred from the pyproject.toml file of the package.",
-    type=click.STRING,
-)
-@pass_pipeline_context
-@click.pass_context
-async def publish(
-    ctx: click.Context,
-    click_pipeline_context: ClickPipelineContext,
-    pypi_token: str,
-    test_pypi: bool,
-    publish_name: Optional[str],
-    publish_version: Optional[str],
-) -> None:
-    context = PyPIPublishContext(
-        is_local=ctx.obj["is_local"],
-        git_branch=ctx.obj["git_branch"],
-        git_revision=ctx.obj["git_revision"],
-        ci_report_bucket=ctx.obj["ci_report_bucket_name"],
-        report_output_prefix=ctx.obj["report_output_prefix"],
-        gha_workflow_run_url=ctx.obj.get("gha_workflow_run_url"),
-        dagger_logs_url=ctx.obj.get("dagger_logs_url"),
-        pipeline_start_timestamp=ctx.obj.get("pipeline_start_timestamp"),
-        ci_context=ctx.obj.get("ci_context"),
-        ci_gcs_credentials=ctx.obj["ci_gcs_credentials"],
-        pypi_token=pypi_token,
-        test_pypi=test_pypi,
-        package_path=ctx.obj["package_path"],
-        build_docker_image=ctx.obj["docker_image"],
-        package_name=publish_name,
-        version=publish_version,
-    )
-    dagger_client = await click_pipeline_context.get_dagger_client(pipeline_name=f"Publish {ctx.obj['package_path']} to PyPI")
-    context.dagger_client = dagger_client
-
-    await PublishToPyPI(context).run()
-
-    return True