fix(airbyte-ci): cleanup unused / empty schema directories after inlining schemas to manifests (#43297)

Co-authored-by: Octavia Squidington III <[email protected]>

Author: natikgadzhi

airbyte-ci/connectors/pipelines/README.md

@@ -773,7 +773,8 @@ E.G.: running Poe tasks on the modified internal packages of the current branch:
 
 | Version | PR                                                         | Description                                                                                                                  |
 | ------- | ---------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------- |
-| 4.28.1  | [#42972](https://github.com/airbytehq/airbyte/pull/42972)  | Add airbyte-enterprise support for format command.                                                                           |
+| 4.28.2  | [#43297](https://github.com/airbytehq/airbyte/pull/43297)  | `migrate-to-inline_schemas` removes unused schema files and empty schema dirs.                                               |
+| 4.28.1  | [#42972](https://github.com/airbytehq/airbyte/pull/42972)  | Add airbyte-enterprise support for format commandi                                                                           |
 | 4.28.0  | [#42849](https://github.com/airbytehq/airbyte/pull/42849)  | Couple selection of strict-encrypt variants (e vice versa)                                                                   |
 | 4.27.0  | [#42574](https://github.com/airbytehq/airbyte/pull/42574)  | Live tests: run from connectors test pipeline for connectors with sandbox connections                                        |
 | 4.26.1  | [#42905](https://github.com/airbytehq/airbyte/pull/42905)  | Rename the docker cache volume to avoid using the corrupted previous volume.                                                 |

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

@@ -34,6 +34,7 @@ class CONNECTOR_TEST_STEP_ID(str, Enum):
     UPDATE_PULL_REQUEST = "up_to_date.pull"
     INLINE_CANDIDATE = "migration_to_inline_schemas.candidate"
     INLINE_MIGRATION = "migration_to_inline_schemas.migration"
+    INLINE_CLEANUP = "migration_to_inline_schemas.cleanup"
     AIRBYTE_LOGGER_CANDIDATE = "migration_to_logging_logger.candidate"
     AIRBYTE_LOGGER_MIGRATION = "migration_to_logging_logger.migration"
     PULL_REQUEST_CREATE = "pull_request.create"

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

@@ -49,7 +49,11 @@ async def _run(self) -> StepResult:
         connector = self.context.connector
         manifest_path = connector.manifest_path
         python_path = connector.python_source_dir_path
-        if connector.language not in [ConnectorLanguage.PYTHON, ConnectorLanguage.LOW_CODE, ConnectorLanguage.MANIFEST_ONLY]:
+        if connector.language not in [
+            ConnectorLanguage.PYTHON,
+            ConnectorLanguage.LOW_CODE,
+            ConnectorLanguage.MANIFEST_ONLY,
+        ]:
             return StepResult(
                 step=self,
                 status=StepStatus.SKIPPED,
@@ -153,7 +157,11 @@ async def _run(self) -> StepResult:
 
         data = read_yaml(manifest_path)
         if "streams" not in data:
-            return StepResult(step=self, status=StepStatus.SKIPPED, stderr="No manifest streams found.")
+            return StepResult(
+                step=self,
+                status=StepStatus.SKIPPED,
+                stderr="No manifest streams found.",
+            )
 
         # find the explit ones and remove or udpate
         json_loaders = _find_json_loaders(data, [])
@@ -196,14 +204,43 @@ async def _run(self) -> StepResult:
             _update_inline_schema(schema_loader, json_streams, stream_name)
 
         write_yaml(data, manifest_path)
-        await format_prettier([manifest_path])
+        await format_prettier([manifest_path], logger=logger)
 
         for json_stream in json_streams.values():
             logger.info(f"     !! JSON schema not found: {json_stream.name}")
 
         return StepResult(step=self, status=StepStatus.SUCCESS)
 
 
+class RemoveUnusedJsonSchamas(Step):
+    context: ConnectorContext
+
+    title = "Cleanup json schemas that are dangling but unused."
+
+    async def _run(self) -> StepResult:
+        connector = self.context.connector
+        connector_path = connector.code_directory
+        manifest_path = connector.manifest_path
+        python_path = connector.python_source_dir_path
+        schemas_path = python_path / SCHEMAS_DIR_NAME
+        logger = self.logger
+
+        manifest = connector.manifest_path.read_text()
+
+        if manifest.find("JsonFileSchemaLoader") != -1:
+            return StepResult(
+                step=self,
+                status=StepStatus.SKIPPED,
+                stderr="Skipping: the manifest is still using JSON Schema loader.",
+            )
+
+        if schemas_path.exists():
+            logger.info(f"    Removing schemnas dir: {schemas_path}")
+            shutil.rmtree(schemas_path)
+
+        return StepResult(step=self, status=StepStatus.SUCCESS)
+
+
 @dataclass
 class JsonStream:
     name: str
@@ -370,7 +407,14 @@ async def run_connector_migrate_to_inline_schemas_pipeline(context: ConnectorCon
 
     steps_to_run: STEP_TREE = []
 
-    steps_to_run.append([StepToRun(id=CONNECTOR_TEST_STEP_ID.INLINE_CANDIDATE, step=CheckIsInlineCandidate(context))])
+    steps_to_run.append(
+        [
+            StepToRun(
+                id=CONNECTOR_TEST_STEP_ID.INLINE_CANDIDATE,
+                step=CheckIsInlineCandidate(context),
+            )
+        ]
+    )
 
     steps_to_run.append(
         [
@@ -382,4 +426,14 @@ async def run_connector_migrate_to_inline_schemas_pipeline(context: ConnectorCon
         ]
     )
 
+    steps_to_run.append(
+        [
+            StepToRun(
+                id=CONNECTOR_TEST_STEP_ID.INLINE_CLEANUP,
+                step=RemoveUnusedJsonSchamas(context),
+                depends_on=[CONNECTOR_TEST_STEP_ID.INLINE_MIGRATION],
+            )
+        ]
+    )
+
     return await run_connector_steps(context, semaphore, steps_to_run, restore_original_state=restore_original_state)

airbyte-ci/connectors/pipelines/pipelines/helpers/connectors/format.py

@@ -1,13 +1,14 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 
+import logging
 import subprocess
 from pathlib import Path
 from typing import List
 
 from pipelines.cli.ensure_repo_root import get_airbyte_repo_path_with_fallback
 
 
-async def format_prettier(files: List[Path]) -> None:
+async def format_prettier(files: List[Path], logger: logging.Logger) -> None:
     if len(files) == 0:
         return
 
@@ -18,13 +19,13 @@ async def format_prettier(files: List[Path]) -> None:
 
     to_format = [str(file) for file in files]
 
-    print(f"       Formatting files: npx prettier --write {' '.join(to_format)}")
+    logger.info(f"       Formatting files: npx prettier --write {' '.join(to_format)}")
     command = ["npx", "prettier", "--config", str(config_path), "--write"] + to_format
     result = subprocess.run(command, capture_output=True, text=True)
     if result.returncode == 0:
-        print("Files formatted successfully.")
+        logger.info("        Files formatted successfully.")
     else:
-        print("Error formatting files.")
+        logger.warn("        Error formatting files.")
 
 
 def verify_formatters() -> None:

airbyte-ci/connectors/pipelines/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "pipelines"
-version = "4.28.1"
+version = "4.28.2"
 description = "Packaged maintained by the connector operations team to perform CI for connectors' pipelines"
 authors = ["Airbyte <[email protected]>"]