airbyte-ci: Add upgrade cdk command

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

@@ -153,6 +153,7 @@ def should_use_remote_secrets(use_remote_secrets: Optional[bool]) -> bool:
         "bump_version": "pipelines.airbyte_ci.connectors.bump_version.commands.bump_version",
         "migrate_to_base_image": "pipelines.airbyte_ci.connectors.migrate_to_base_image.commands.migrate_to_base_image",
         "upgrade_base_image": "pipelines.airbyte_ci.connectors.upgrade_base_image.commands.upgrade_base_image",
+        "upgrade_cdk": "pipelines.airbyte_ci.connectors.upgrade_cdk.commands.bump_version",
     },
 )
 @click.option(

airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/upgrade_cdk/__init__.py

@@ -0,0 +1,3 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#

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

@@ -0,0 +1,55 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+import asyncclick as click
+from pipelines.airbyte_ci.connectors.upgrade_cdk.pipeline import run_connector_cdk_upgrade_pipeline
+from pipelines.airbyte_ci.connectors.context import ConnectorContext
+from pipelines.airbyte_ci.connectors.pipeline import run_connectors_pipelines
+from pipelines.cli.dagger_pipeline_command import DaggerPipelineCommand
+
+
+@click.command(cls=DaggerPipelineCommand, short_help="Upgrade CDK version")
+@click.argument("target-version", type=str)
+@click.pass_context
+async def bump_version(
+    ctx: click.Context,
+    target_version: str,
+) -> bool:
+    """Upgrade CDK version"""
+
+    connectors_contexts = [
+        ConnectorContext(
+            pipeline_name=f"Upgrade CDK version of connector {connector.technical_name}",
+            connector=connector,
+            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"],
+            use_remote_secrets=ctx.obj["use_remote_secrets"],
+            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"],
+            ci_git_user=ctx.obj["ci_git_user"],
+            ci_github_access_token=ctx.obj["ci_github_access_token"],
+            enable_report_auto_open=False,
+            s3_build_cache_access_key_id=ctx.obj.get("s3_build_cache_access_key_id"),
+            s3_build_cache_secret_key=ctx.obj.get("s3_build_cache_secret_key"),
+        )
+        for connector in ctx.obj["selected_connectors_with_modified_files"]
+    ]
+
+    await run_connectors_pipelines(
+        connectors_contexts,
+        run_connector_cdk_upgrade_pipeline,
+        "Upgrade CDK version pipeline",
+        ctx.obj["concurrency"],
+        ctx.obj["dagger_logs_path"],
+        ctx.obj["execute_timeout"],
+        target_version,
+    )
+
+    return True

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

@@ -0,0 +1,93 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+import re
+
+from dagger import Container
+from pipelines.airbyte_ci.connectors.context import ConnectorContext
+from pipelines.airbyte_ci.connectors.reports import ConnectorReport
+from pipelines.helpers import git
+from pipelines.models.steps import Step, StepResult, StepStatus
+
+
+class SetCDKVersion(Step):
+    title = "Set CDK Version"
+
+    def __init__(
+        self,
+        context: ConnectorContext,
+        repo_dir: Container,
+        new_version: str,
+    ):
+        super().__init__(context)
+        self.repo_dir = repo_dir
+        self.new_version = new_version
+
+    async def _run(self) -> StepResult:
+        context: ConnectorContext = self.context
+        setup_py_path = context.connector.code_directory / "setup.py"
+        if not setup_py_path.exists():
+            return StepResult(
+                self,
+                StepStatus.SKIPPED,
+                stdout="Connector does not have a setup.py file.",
+                output_artifact=self.repo_dir,
+            )
+        try:
+            updated_setup_py = self.update_cdk_version(setup_py_path.read_text())
+        except Exception as e:
+            return StepResult(
+                self,
+                StepStatus.FAILURE,
+                stdout=f"Could not add changelog entry: {e}",
+                output_artifact=self.container_with_airbyte_repo,
+            )
+        updated_repo_dir = self.repo_dir.with_new_file(str(setup_py_path), updated_setup_py)
+        return StepResult(
+            self,
+            StepStatus.SUCCESS,
+            stdout=f"Updated CDK version to {self.new_version}",
+            output_artifact=updated_repo_dir,
+        )
+
+    def update_cdk_version(self, og_setup_py_content: str) -> str:
+        airbyte_cdk_dependency = re.search(
+            r"airbyte-cdk(?P<extra>\[[a-zA-Z0-9-]*\])?(?P<version>[<>=!~]+[0-9]*\.[0-9]*\.[0-9]*)?", og_setup_py_content
+        )
+        # If there is no airbyte-cdk dependency, add the version
+        if airbyte_cdk_dependency is not None:
+            new_version = f"airbyte-cdk{airbyte_cdk_dependency.group('extra') or ''}>={self.new_version}"
+            return og_setup_py_content.replace(airbyte_cdk_dependency.group(), new_version)
+        else:
+            raise ValueError("Could not find airbyte-cdk dependency in setup.py")
+
+async def run_connector_cdk_upgrade_pipeline(
+    context: ConnectorContext,
+    semaphore,
+    target_version: str,
+) -> ConnectorReport:
+    """Run a pipeline to upgrade the CDK version for a single connector.
+
+    Args:
+        context (ConnectorContext): The initialized connector context.
+
+    Returns:
+        ConnectorReport: The reports holding the base image version upgrade results.
+    """
+    async with semaphore:
+        steps_results = []
+        async with context:
+            og_repo_dir = await context.get_repo_dir()
+
+            set_cdk_version = SetCDKVersion(
+                context,
+                og_repo_dir,
+                target_version,
+            )
+            set_cdk_version_result = await set_cdk_version.run()
+            steps_results.append(set_cdk_version_result)
+            final_repo_dir = set_cdk_version_result.output_artifact
+            await og_repo_dir.diff(final_repo_dir).export(str(git.get_git_repo_path()))
+            context.report = ConnectorReport(context, steps_results, name="CONNECTOR VERSION CDK UPGRADE RESULTS")
+    return context.report

airbyte-ci/connectors/pipelines/tests/test_upgrade_cdk.py

@@ -0,0 +1,49 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+import json
+import random
+from typing import List
+
+import anyio
+import pytest
+from connector_ops.utils import Connector, ConnectorLanguage
+from pipelines.airbyte_ci.connectors.publish import pipeline as publish_pipeline
+from pipelines.airbyte_ci.connectors.upgrade_cdk import pipeline as upgrade_cdk_pipeline
+from pipelines.models.steps import StepStatus
+from pipelines.airbyte_ci.connectors.context import ConnectorContext
+
+pytestmark = [
+    pytest.mark.anyio,
+]
+
+@pytest.fixture
+def connector_with_poetry():
+    return Connector("destination-duckdb")
+
+@pytest.fixture
+def context_for_connector_with_poetry(connector_with_poetry, dagger_client, current_platform):
+    context = ConnectorContext(
+        pipeline_name="test unit tests",
+        connector=connector_with_poetry,
+        git_branch="test",
+        git_revision="test",
+        report_output_prefix="test",
+        is_local=True,
+        use_remote_secrets=True,
+        targeted_platforms=[current_platform],
+    )
+    context.dagger_client = dagger_client
+    context.connector_secrets = {}
+    return context
+
+async def test_run_connector_cdk_upgrade_pipeline(context_for_connector_with_poetry):
+    # og_repo_dir = await context.get_repo_dir()
+    # report = await upgrade_cdk_pipeline.run_connector_cdk_upgrade_pipeline(context_for_connector_with_poetry, semaphore, "6.6.6")
+    og_repo_dir = await context_for_connector_with_poetry.get_repo_dir()
+    step = upgrade_cdk_pipeline.SetCDKVersion(context_for_connector_with_poetry, og_repo_dir, "6.6.6")
+    step_result = await step.run()
+
+    # print(report.steps_results[-1])
+    print(step_result)