feat: expose operations in public Pipeline API

kpops/api/__init__.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Iterator
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -14,33 +13,20 @@
 from kpops.component_handlers.topic.handler import TopicHandler
 from kpops.component_handlers.topic.kafka_rest import KafkaRest
 from kpops.config import KpopsConfig
-from kpops.core.exception import KpopsException
 from kpops.core.operation import OperationMode
 from kpops.core.registry import Registry
-from kpops.manifests.kubernetes import KubernetesManifest
 from kpops.pipeline import (
     Pipeline,
     PipelineGenerator,
 )
 from kpops.utils.cli_commands import init_project
-from kpops.utils.logging import log, log_action, log_kpops_exception
+from kpops.utils.logging import log
 
 if TYPE_CHECKING:
-    from collections.abc import Awaitable
+    from collections.abc import Iterator
 
-    from kpops.components.base_components.pipeline_component import PipelineComponent
     from kpops.config import KpopsConfig
-
-
-async def _run_component(
-    action: str, component: PipelineComponent, operation: Awaitable[None]
-) -> None:
-    log_action(action, component)
-    try:
-        await operation
-    except KpopsException as e:
-        log_kpops_exception(e)
-        raise
+    from kpops.manifests.kubernetes import KubernetesManifest
 
 
 def generate(
@@ -106,9 +92,7 @@ def manifest_deploy(
         verbose=verbose,
         operation_mode=operation_mode,
     )
-    for component in pipeline.components:
-        resource = component.manifest_deploy()
-        yield resource
+    return pipeline.manifest_deploy()
 
 
 def manifest_destroy(
@@ -131,9 +115,7 @@ def manifest_destroy(
         verbose=verbose,
         operation_mode=operation_mode,
     )
-    for component in pipeline.components:
-        resource = component.manifest_destroy()
-        yield resource
+    return pipeline.manifest_destroy()
 
 
 def manifest_reset(
@@ -156,9 +138,7 @@ def manifest_reset(
         verbose=verbose,
         operation_mode=operation_mode,
     )
-    for component in pipeline.components:
-        resource = component.manifest_reset()
-        yield resource
+    return pipeline.manifest_reset()
 
 
 def manifest_clean(
@@ -181,9 +161,7 @@ def manifest_clean(
         verbose=verbose,
         operation_mode=operation_mode,
     )
-    for component in pipeline.components:
-        resource = component.manifest_clean()
-        yield resource
+    return pipeline.manifest_clean()
 
 
 def deploy(
@@ -218,19 +196,7 @@ def deploy(
         environment=environment,
         verbose=verbose,
     )
-
-    async def deploy_runner(component: PipelineComponent) -> None:
-        await _run_component("Deploy", component, component.deploy(dry_run))
-
-    async def async_deploy() -> None:
-        if parallel:
-            pipeline_tasks = pipeline.build_execution_graph(deploy_runner)
-            await pipeline_tasks
-        else:
-            for component in pipeline.components:
-                await deploy_runner(component)
-
-    asyncio.run(async_deploy())
+    asyncio.run(pipeline.deploy(dry_run, parallel))
 
 
 def destroy(
@@ -265,21 +231,7 @@ def destroy(
         environment=environment,
         verbose=verbose,
     )
-
-    async def destroy_runner(component: PipelineComponent) -> None:
-        await _run_component("Destroy", component, component.destroy(dry_run))
-
-    async def async_destroy() -> None:
-        if parallel:
-            pipeline_tasks = pipeline.build_execution_graph(
-                destroy_runner, reverse=True
-            )
-            await pipeline_tasks
-        else:
-            for component in reversed(pipeline.components):
-                await destroy_runner(component)
-
-    asyncio.run(async_destroy())
+    asyncio.run(pipeline.destroy(dry_run, parallel))
 
 
 def reset(
@@ -314,19 +266,7 @@ def reset(
         environment=environment,
         verbose=verbose,
     )
-
-    async def reset_runner(component: PipelineComponent) -> None:
-        await _run_component("Reset", component, component.reset(dry_run))
-
-    async def async_reset() -> None:
-        if parallel:
-            pipeline_tasks = pipeline.build_execution_graph(reset_runner, reverse=True)
-            await pipeline_tasks
-        else:
-            for component in reversed(pipeline.components):
-                await reset_runner(component)
-
-    asyncio.run(async_reset())
+    asyncio.run(pipeline.reset(dry_run, parallel))
 
 
 def clean(
@@ -361,19 +301,7 @@ def clean(
         environment=environment,
         verbose=verbose,
     )
-
-    async def clean_runner(component: PipelineComponent) -> None:
-        await _run_component("Clean", component, component.clean(dry_run))
-
-    async def async_clean() -> None:
-        if parallel:
-            pipeline_tasks = pipeline.build_execution_graph(clean_runner, reverse=True)
-            await pipeline_tasks
-        else:
-            for component in reversed(pipeline.components):
-                await clean_runner(component)
-
-    asyncio.run(async_clean())
+    asyncio.run(pipeline.clean(dry_run, parallel))
 
 
 def init(

kpops/pipeline/__init__.py

@@ -15,23 +15,36 @@
 
 from kpops.component_handlers import ComponentHandlers
 from kpops.components.base_components.pipeline_component import PipelineComponent
-from kpops.core.exception import ParsingException, ValidationError
+from kpops.core.exception import KpopsException, ParsingException, ValidationError
 from kpops.core.registry import Registry
 from kpops.utils.dict_ops import update_nested_pair
 from kpops.utils.environment import ENV, PIPELINE_PATH
+from kpops.utils.logging import log_action, log_kpops_exception
 from kpops.utils.yaml import CustomSafeDumper, load_yaml_file
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Coroutine, Iterator
     from pathlib import Path
 
     from kpops.config import KpopsConfig
+    from kpops.manifests.kubernetes import KubernetesManifest
 
 log = structlog.get_logger("PipelineGenerator")
 
 ComponentFilterPredicate: TypeAlias = Callable[[PipelineComponent], bool]
 
 
+async def _run_component(
+    action: str, component: PipelineComponent, operation: Awaitable[None]
+) -> None:
+    log_action(action, component)
+    try:
+        await operation
+    except KpopsException as e:
+        log_kpops_exception(e)
+        raise
+
+
 @dataclass
 class Pipeline:
     """Pipeline representation."""
@@ -143,6 +156,85 @@ async def run_graph_layers(
 
         return run_graph_layers(sorted_layers)
 
+    async def deploy(self, dry_run: bool, parallel: bool = False) -> None:
+        """Deploy pipeline steps.
+
+        :param dry_run: Whether to dry run the command or execute it.
+        :param parallel: Enable or disable parallel execution of pipeline steps.
+        """
+        await self._run_action(
+            "Deploy",
+            lambda component: component.deploy(dry_run),
+            parallel,
+            reverse=False,
+        )
+
+    async def destroy(self, dry_run: bool, parallel: bool = False) -> None:
+        """Destroy pipeline steps.
+
+        :param dry_run: Whether to dry run the command or execute it.
+        :param parallel: Enable or disable parallel execution of pipeline steps.
+        """
+        await self._run_action(
+            "Destroy",
+            lambda component: component.destroy(dry_run),
+            parallel,
+            reverse=True,
+        )
+
+    async def reset(self, dry_run: bool, parallel: bool = False) -> None:
+        """Reset pipeline steps.
+
+        :param dry_run: Whether to dry run the command or execute it.
+        :param parallel: Enable or disable parallel execution of pipeline steps.
+        """
+        await self._run_action(
+            "Reset", lambda component: component.reset(dry_run), parallel, reverse=True
+        )
+
+    async def clean(self, dry_run: bool, parallel: bool = False) -> None:
+        """Clean pipeline steps.
+
+        :param dry_run: Whether to dry run the command or execute it.
+        :param parallel: Enable or disable parallel execution of pipeline steps.
+        """
+        await self._run_action(
+            "Clean", lambda component: component.clean(dry_run), parallel, reverse=True
+        )
+
+    def manifest_deploy(self) -> Iterator[tuple[KubernetesManifest, ...]]:
+        for component in self.components:
+            yield component.manifest_deploy()
+
+    def manifest_destroy(self) -> Iterator[tuple[KubernetesManifest, ...]]:
+        for component in self.components:
+            yield component.manifest_destroy()
+
+    def manifest_reset(self) -> Iterator[tuple[KubernetesManifest, ...]]:
+        for component in self.components:
+            yield component.manifest_reset()
+
+    def manifest_clean(self) -> Iterator[tuple[KubernetesManifest, ...]]:
+        for component in self.components:
+            yield component.manifest_clean()
+
+    async def _run_action(
+        self,
+        action_name: str,
+        component_action: Callable[[PipelineComponent], Coroutine[Any, Any, None]],
+        parallel: bool,
+        reverse: bool,
+    ) -> None:
+        async def runner(component: PipelineComponent) -> None:
+            await _run_component(action_name, component, component_action(component))
+
+        if parallel:
+            await self.build_execution_graph(runner, reverse=reverse)
+        else:
+            components = reversed(self.components) if reverse else self.components
+            for component in components:
+                await runner(component)
+
     def __getitem__(self, component_id: str) -> PipelineComponent:
         try:
             return self._component_index[component_id]

tests/api/test_handlers.py

@@ -30,6 +30,7 @@ def handlers() -> Generator[ComponentHandlers, None, None]:
     ComponentHandlers._instance = None
 
 
+@pytest.mark.usefixtures("clear_handlers")
 def test_global_handlers_not_initialized() -> None:
     with pytest.raises(
         RuntimeError, match="ComponentHandlers has not been initialized"

tests/cli/test_init.py

@@ -23,7 +23,7 @@ def test_create_config(tmp_path: Path) -> None:
     assert len(opt_conf.read_text()) > len(req_conf.read_text())
 
 
-@pytest.mark.usefixtures("mock_env", "load_yaml_file_clear_cache", "clear_kpops_config")
+@pytest.mark.usefixtures("mock_env", "load_yaml_file_clear_cache", "clear_config")
 def test_init_project_exclude_optional(tmp_path: Path, snapshot: Snapshot) -> None:
     req_path = tmp_path / "req"
     req_path.mkdir()

tests/components/conftest.py

@@ -1,29 +1,19 @@
-from unittest import mock
+from pathlib import Path
 
 import pytest
 
 from kpops.component_handlers import ComponentHandlers
-from kpops.config import KpopsConfig, TopicNameConfig, set_config
+from kpops.config import KpopsConfig
 from tests.components import PIPELINE_BASE_DIR
 
 
-@pytest.fixture(autouse=True, scope="module")
-def config() -> None:
-    config = KpopsConfig(
-        topic_name_config=TopicNameConfig(
-            default_error_topic_name="${component.type}-error-topic",
-            default_output_topic_name="${component.type}-output-topic",
-        ),
-        kafka_brokers="broker:9092",
-        pipeline_base_dir=PIPELINE_BASE_DIR,
-    )
-    set_config(config)
+@pytest.fixture(scope="module")
+def pipeline_base_dir() -> Path:
+    return PIPELINE_BASE_DIR
 
 
 @pytest.fixture(autouse=True, scope="module")
-def handlers() -> None:
-    ComponentHandlers(
-        schema_handler=mock.AsyncMock(),
-        connector_handler=mock.AsyncMock(),
-        topic_handler=mock.AsyncMock(),
-    )
+def _apply_config_and_handlers(
+    config: KpopsConfig, handlers: ComponentHandlers
+) -> None:
+    pass