Python API

kpops.api

clean

clean(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None

Clean pipeline steps.

PARAMETER	DESCRIPTION
pipeline_path	Path to pipeline definition yaml file. TYPE: Path
dotenv	Paths to dotenv files. TYPE: list[Path] | None DEFAULT: None
config	Path to the dir containing config.yaml files. TYPE: Path | None DEFAULT: None
steps	Set of steps (components) to apply the command on. TYPE: set[str] | None DEFAULT: None
filter_type	Whether steps should include/exclude the steps. TYPE: FilterType DEFAULT: INCLUDE
dry_run	Whether to dry run the command or execute it. TYPE: bool DEFAULT: True
environment	The environment to generate and deploy the pipeline to. TYPE: str | None DEFAULT: None
verbose	Enable verbose printing. TYPE: bool DEFAULT: True
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/api/operations.py

def clean(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None:
    """Clean pipeline steps.

    :param pipeline_path: Path to pipeline definition yaml file.
    :param dotenv: Paths to dotenv files.
    :param config: Path to the dir containing config.yaml files.
    :param steps: Set of steps (components) to apply the command on.
    :param filter_type: Whether `steps` should include/exclude the steps.
    :param dry_run: Whether to dry run the command or execute it.
    :param environment: The environment to generate and deploy the pipeline to.
    :param verbose: Enable verbose printing.
    :param parallel: Enable or disable parallel execution of pipeline steps.
    """
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
    )
    asyncio.run(pipeline.clean(dry_run, parallel))

deploy

deploy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None

Deploy pipeline steps.

PARAMETER	DESCRIPTION
pipeline_path	Path to pipeline definition yaml file. TYPE: Path
dotenv	Paths to dotenv files. TYPE: list[Path] | None DEFAULT: None
config	Path to the dir containing config.yaml files. TYPE: Path | None DEFAULT: None
steps	Set of steps (components) to apply the command on. TYPE: set[str] | None DEFAULT: None
filter_type	Whether steps should include/exclude the steps. TYPE: FilterType DEFAULT: INCLUDE
dry_run	Whether to dry run the command or execute it. TYPE: bool DEFAULT: True
environment	The environment to generate and deploy the pipeline to. TYPE: str | None DEFAULT: None
verbose	Enable verbose printing. TYPE: bool DEFAULT: True
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/api/operations.py

def deploy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None:
    """Deploy pipeline steps.

    :param pipeline_path: Path to pipeline definition yaml file.
    :param dotenv: Paths to dotenv files.
    :param config: Path to the dir containing config.yaml files.
    :param steps: Set of steps (components) to apply the command on.
    :param filter_type: Whether `steps` should include/exclude the steps.
    :param dry_run: Whether to dry run the command or execute it.
    :param environment: The environment to generate and deploy the pipeline to.
    :param verbose: Enable verbose printing.
    :param parallel: Enable or disable parallel execution of pipeline steps.
    """
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
    )
    asyncio.run(pipeline.deploy(dry_run, parallel))

destroy

destroy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None

Destroy pipeline steps.

PARAMETER	DESCRIPTION
pipeline_path	Path to pipeline definition yaml file. TYPE: Path
dotenv	Paths to dotenv files. TYPE: list[Path] | None DEFAULT: None
config	Path to the dir containing config.yaml files. TYPE: Path | None DEFAULT: None
steps	Set of steps (components) to apply the command on. TYPE: set[str] | None DEFAULT: None
filter_type	Whether steps should include/exclude the steps. TYPE: FilterType DEFAULT: INCLUDE
dry_run	Whether to dry run the command or execute it. TYPE: bool DEFAULT: True
environment	The environment to generate and deploy the pipeline to. TYPE: str | None DEFAULT: None
verbose	Enable verbose printing. TYPE: bool DEFAULT: True
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/api/operations.py

def destroy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None:
    """Destroy pipeline steps.

    :param pipeline_path: Path to pipeline definition yaml file.
    :param dotenv: Paths to dotenv files.
    :param config: Path to the dir containing config.yaml files.
    :param steps: Set of steps (components) to apply the command on.
    :param filter_type: Whether `steps` should include/exclude the steps.
    :param dry_run: Whether to dry run the command or execute it.
    :param environment: The environment to generate and deploy the pipeline to.
    :param verbose: Enable verbose printing.
    :param parallel: Enable or disable parallel execution of pipeline steps.
    """
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
    )
    asyncio.run(pipeline.destroy(dry_run, parallel))

generate

generate(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = False,
    operation_mode: OperationMode = OperationMode.MANAGED,
) -> Pipeline

Generate enriched pipeline representation.

PARAMETER	DESCRIPTION
pipeline_path	Path to pipeline definition yaml file. TYPE: Path
dotenv	Paths to dotenv files. TYPE: list[Path] | None DEFAULT: None
config	Path to the dir containing config.yaml files. TYPE: Path | None DEFAULT: None
steps	Set of steps (components) to apply the command on. TYPE: set[str] | None DEFAULT: None
filter_type	Whether steps should include/exclude the steps. TYPE: FilterType DEFAULT: INCLUDE
environment	The environment to generate and deploy the pipeline to. TYPE: str | None DEFAULT: None
verbose	Enable verbose printing. TYPE: bool DEFAULT: False
operation_mode	How KPOps should operate. TYPE: OperationMode DEFAULT: MANAGED

RETURNS	DESCRIPTION
Pipeline	Generated Pipeline object.

Source code in kpops/api/operations.py

def generate(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = False,
    operation_mode: OperationMode = OperationMode.MANAGED,
) -> Pipeline:
    """Generate enriched pipeline representation.

    :param pipeline_path: Path to pipeline definition yaml file.
    :param dotenv: Paths to dotenv files.
    :param config: Path to the dir containing config.yaml files.
    :param steps: Set of steps (components) to apply the command on.
    :param filter_type: Whether `steps` should include/exclude the steps.
    :param environment: The environment to generate and deploy the pipeline to.
    :param verbose: Enable verbose printing.
    :param operation_mode: How KPOps should operate.
    :return: Generated `Pipeline` object.
    """
    kpops_config = KpopsConfig.create(
        config, dotenv, environment, verbose, operation_mode
    )
    pipeline = _create_pipeline(pipeline_path, kpops_config, environment)
    log.info("Picked up pipeline", pipeline=pipeline_path.parent.name)
    if steps:
        component_names = steps
        log.debug(
            "KPOPS_PIPELINE_STEPS is defined",
            steps=component_names,
            filter_type=filter_type.value,
        )

        predicate = filter_type.create_default_step_names_filter_predicate(
            component_names
        )
        pipeline.filter(predicate)
        log.info("Filtered pipeline", steps=pipeline.step_names)
    return pipeline

init

init(
    path: Path, config_include_optional: bool = False
) -> None

Initiate a default empty project.

PARAMETER	DESCRIPTION
path	Directory in which the project should be initiated. TYPE: Path
config_include_optional	Whether to include non-required settings in the generated config file. TYPE: bool DEFAULT: False

Source code in kpops/api/operations.py

def init(
    path: Path,
    config_include_optional: bool = False,
) -> None:
    """Initiate a default empty project.

    :param path: Directory in which the project should be initiated.
    :param config_include_optional: Whether to include non-required settings
        in the generated config file.
    """
    if not path.exists():
        path.mkdir(parents=False)
    elif next(path.iterdir(), False):
        log.warning("Please provide a path to an empty directory.")
        return
    init_project(path, config_include_optional)

manifest_clean

manifest_clean(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]

Source code in kpops/api/operations.py

def manifest_clean(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]:
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
        operation_mode=operation_mode,
    )
    yield from pipeline.manifest_clean()

manifest_deploy

manifest_deploy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]

Source code in kpops/api/operations.py

def manifest_deploy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]:
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
        operation_mode=operation_mode,
    )
    yield from pipeline.manifest_deploy()

manifest_destroy

manifest_destroy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]

Source code in kpops/api/operations.py

def manifest_destroy(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]:
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
        operation_mode=operation_mode,
    )
    yield from pipeline.manifest_destroy()

manifest_reset

manifest_reset(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]

Source code in kpops/api/operations.py

def manifest_reset(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    verbose: bool = True,
    operation_mode: OperationMode = OperationMode.MANIFEST,
) -> Iterator[tuple[KubernetesManifest, ...]]:
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
        operation_mode=operation_mode,
    )
    yield from pipeline.manifest_reset()

reset

reset(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None

Reset pipeline steps.

PARAMETER	DESCRIPTION
pipeline_path	Path to pipeline definition yaml file. TYPE: Path
dotenv	Paths to dotenv files. TYPE: list[Path] | None DEFAULT: None
config	Path to the dir containing config.yaml files. TYPE: Path | None DEFAULT: None
steps	Set of steps (components) to apply the command on. TYPE: set[str] | None DEFAULT: None
filter_type	Whether steps should include/exclude the steps. TYPE: FilterType DEFAULT: INCLUDE
dry_run	Whether to dry run the command or execute it. TYPE: bool DEFAULT: True
environment	The environment to generate and deploy the pipeline to. TYPE: str | None DEFAULT: None
verbose	Enable verbose printing. TYPE: bool DEFAULT: True
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/api/operations.py

def reset(
    pipeline_path: Path,
    dotenv: list[Path] | None = None,
    config: Path | None = None,
    steps: set[str] | None = None,
    filter_type: FilterType = FilterType.INCLUDE,
    environment: str | None = None,
    dry_run: bool = True,
    verbose: bool = True,
    parallel: bool = False,
) -> None:
    """Reset pipeline steps.

    :param pipeline_path: Path to pipeline definition yaml file.
    :param dotenv: Paths to dotenv files.
    :param config: Path to the dir containing config.yaml files.
    :param steps: Set of steps (components) to apply the command on.
    :param filter_type: Whether `steps` should include/exclude the steps.
    :param dry_run: Whether to dry run the command or execute it.
    :param environment: The environment to generate and deploy the pipeline to.
    :param verbose: Enable verbose printing.
    :param parallel: Enable or disable parallel execution of pipeline steps.
    """
    pipeline = generate(
        pipeline_path=pipeline_path,
        dotenv=dotenv,
        config=config,
        steps=steps,
        filter_type=filter_type,
        environment=environment,
        verbose=verbose,
    )
    asyncio.run(pipeline.reset(dry_run, parallel))

kpops.pipeline.Pipeline dataclass

Pipeline representation.

Source code in kpops/pipeline.py

@dataclass
class Pipeline:
    """Pipeline representation."""

    _component_index: dict[str, PipelineComponent] = field(default_factory=dict)
    _graph: rx.PyDiGraph[str, None] = field(default_factory=rx.PyDiGraph)
    _node_index: dict[str, int] = field(default_factory=dict)

    @property
    def step_names(self) -> list[str]:
        return [step.name for step in self.components]

    @computed_field(title="Components")
    @property
    def components(self) -> list[SerializeAsAny[PipelineComponent]]:
        return list(self._component_index.values())

    @property
    def last(self) -> PipelineComponent:
        return self.components[-1]

    def add(self, component: PipelineComponent) -> None:
        if self._component_index.get(component.id) is not None:
            msg = (
                f"Pipeline steps must have unique id, '{component.id}' already exists."
            )
            raise ValidationError(msg)
        self._component_index[component.id] = component
        self.__add_to_graph(component)

    def remove(self, component_id: str) -> None:
        self._component_index.pop(component_id)

    def get(self, component_id: str) -> PipelineComponent | None:
        return self._component_index.get(component_id)

    def find(self, predicate: ComponentFilterPredicate) -> Iterator[PipelineComponent]:
        """Find pipeline components matching a custom predicate.

        :param predicate: Filter function,
            returns boolean value whether the component should be kept or removed
        :returns: Iterator of components matching the predicate
        """
        for component in self.components:
            if predicate(component):
                yield component

    def filter(self, predicate: ComponentFilterPredicate) -> None:
        """Filter pipeline components using a custom predicate.

        :param predicate: Filter function,
            returns boolean value whether the component should be kept or removed
        """
        for component in self.components:
            # filter out components not matching the predicate
            if not predicate(component):
                self.remove(component.id)

    def validate(self) -> None:
        self.__validate_graph()

    def generate(self) -> list[dict[str, Any]]:
        return [component.generate() for component in self.components]

    def to_yaml(self) -> str:
        return yaml.dump(self.generate(), sort_keys=False, Dumper=CustomSafeDumper)

    def build_execution_graph(
        self,
        runner: Callable[[PipelineComponent], Coroutine[Any, Any, None]],
        /,
        reverse: bool = False,
    ) -> Awaitable[None]:
        async def run_layer_parallel(
            components: list[PipelineComponent],
        ) -> None:
            tasks: list[asyncio.Task[None]] = []
            for component in components:
                tasks.append(asyncio.create_task(runner(component)))
            await asyncio.gather(*tasks)

        async def run_graph_layers(
            pending_layers: list[list[PipelineComponent]],
        ) -> None:
            for layer_components in pending_layers:
                await run_layer_parallel(layer_components)

        graph = self._graph.copy()

        # We add an extra node to the graph, connecting all the leaf nodes to it
        # in that way we make this node the root of the graph, avoiding backtracking
        root_node = graph.add_node("root_node_bfs")

        for node in graph.node_indices():
            if node != root_node and not list(graph.predecessors(node)):
                graph.add_edge(root_node, node, None)

        layers_graph = list(rx.bfs_layers(graph, [root_node]))

        sorted_layers: list[list[PipelineComponent]] = []
        for layer in layers_graph[1:]:
            if parallel_components := self.__get_parallel_components_from(
                [graph[idx] for idx in layer]
            ):
                sorted_layers.append(parallel_components)

        if reverse:
            sorted_layers.reverse()

        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]
        except KeyError as exc:
            msg = f"Component {component_id} not found"
            raise ValueError(msg) from exc

    def __bool__(self) -> bool:
        return bool(self._component_index)

    def __iter__(self) -> Iterator[PipelineComponent]:
        yield from self._component_index.values()

    def __len__(self) -> int:
        return len(self.components)

    def __get_or_add_node(self, node_id: str) -> int:
        if node_id not in self._node_index:
            self._node_index[node_id] = self._graph.add_node(node_id)
        return self._node_index[node_id]

    def __add_to_graph(self, component: PipelineComponent) -> None:
        node = self.__get_or_add_node(component.id)

        for input_topic in component.inputs:
            self.__add_input(input_topic.id, node)

        for output_topic in component.outputs:
            self.__add_output(output_topic.id, node)

    def __add_output(self, topic_id: str, source: int) -> None:
        topic = self.__get_or_add_node(topic_id)
        self._graph.add_edge(source, topic, None)

    def __add_input(self, topic_id: str, target: int) -> None:
        topic = self.__get_or_add_node(topic_id)
        self._graph.add_edge(topic, target, None)

    def __get_parallel_components_from(
        self, layer: list[str]
    ) -> list[PipelineComponent]:
        def gen_parallel_components() -> Iterator[PipelineComponent]:
            for node_in_layer in layer:
                # check if component, skip topics
                if (component := self._component_index.get(node_in_layer)) is not None:
                    yield component

        return list(gen_parallel_components())

    def __validate_graph(self) -> None:
        if not rx.is_directed_acyclic_graph(self._graph):
            msg = "Pipeline is not a valid DAG."
            raise ValueError(msg)

components property

components: list[SerializeAsAny[PipelineComponent]]

last property

last: PipelineComponent

step_names property

step_names: list[str]

add

add(component: PipelineComponent) -> None

Source code in kpops/pipeline.py

def add(self, component: PipelineComponent) -> None:
    if self._component_index.get(component.id) is not None:
        msg = (
            f"Pipeline steps must have unique id, '{component.id}' already exists."
        )
        raise ValidationError(msg)
    self._component_index[component.id] = component
    self.__add_to_graph(component)

build_execution_graph

build_execution_graph(
    runner: Callable[
        [PipelineComponent], Coroutine[Any, Any, None]
    ],
    /,
    reverse: bool = False,
) -> Awaitable[None]

Source code in kpops/pipeline.py

def build_execution_graph(
    self,
    runner: Callable[[PipelineComponent], Coroutine[Any, Any, None]],
    /,
    reverse: bool = False,
) -> Awaitable[None]:
    async def run_layer_parallel(
        components: list[PipelineComponent],
    ) -> None:
        tasks: list[asyncio.Task[None]] = []
        for component in components:
            tasks.append(asyncio.create_task(runner(component)))
        await asyncio.gather(*tasks)

    async def run_graph_layers(
        pending_layers: list[list[PipelineComponent]],
    ) -> None:
        for layer_components in pending_layers:
            await run_layer_parallel(layer_components)

    graph = self._graph.copy()

    # We add an extra node to the graph, connecting all the leaf nodes to it
    # in that way we make this node the root of the graph, avoiding backtracking
    root_node = graph.add_node("root_node_bfs")

    for node in graph.node_indices():
        if node != root_node and not list(graph.predecessors(node)):
            graph.add_edge(root_node, node, None)

    layers_graph = list(rx.bfs_layers(graph, [root_node]))

    sorted_layers: list[list[PipelineComponent]] = []
    for layer in layers_graph[1:]:
        if parallel_components := self.__get_parallel_components_from(
            [graph[idx] for idx in layer]
        ):
            sorted_layers.append(parallel_components)

    if reverse:
        sorted_layers.reverse()

    return run_graph_layers(sorted_layers)

clean async

clean(dry_run: bool, parallel: bool = False) -> None

Clean pipeline steps.

PARAMETER	DESCRIPTION
dry_run	Whether to dry run the command or execute it. TYPE: bool
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/pipeline.py

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
    )

deploy async

deploy(dry_run: bool, parallel: bool = False) -> None

Deploy pipeline steps.

PARAMETER	DESCRIPTION
dry_run	Whether to dry run the command or execute it. TYPE: bool
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/pipeline.py

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,
    )

destroy async

destroy(dry_run: bool, parallel: bool = False) -> None

Destroy pipeline steps.

PARAMETER	DESCRIPTION
dry_run	Whether to dry run the command or execute it. TYPE: bool
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/pipeline.py

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,
    )

filter

filter(predicate: ComponentFilterPredicate) -> None

Filter pipeline components using a custom predicate.

PARAMETER	DESCRIPTION
predicate	Filter function, returns boolean value whether the component should be kept or removed TYPE: ComponentFilterPredicate

Source code in kpops/pipeline.py

def filter(self, predicate: ComponentFilterPredicate) -> None:
    """Filter pipeline components using a custom predicate.

    :param predicate: Filter function,
        returns boolean value whether the component should be kept or removed
    """
    for component in self.components:
        # filter out components not matching the predicate
        if not predicate(component):
            self.remove(component.id)

find

find(
    predicate: ComponentFilterPredicate,
) -> Iterator[PipelineComponent]

Find pipeline components matching a custom predicate.

PARAMETER	DESCRIPTION
predicate	Filter function, returns boolean value whether the component should be kept or removed TYPE: ComponentFilterPredicate

RETURNS	DESCRIPTION
Iterator[PipelineComponent]	Iterator of components matching the predicate

Source code in kpops/pipeline.py

def find(self, predicate: ComponentFilterPredicate) -> Iterator[PipelineComponent]:
    """Find pipeline components matching a custom predicate.

    :param predicate: Filter function,
        returns boolean value whether the component should be kept or removed
    :returns: Iterator of components matching the predicate
    """
    for component in self.components:
        if predicate(component):
            yield component

generate

generate() -> list[dict[str, Any]]

Source code in kpops/pipeline.py

def generate(self) -> list[dict[str, Any]]:
    return [component.generate() for component in self.components]

get

get(component_id: str) -> PipelineComponent | None

Source code in kpops/pipeline.py

def get(self, component_id: str) -> PipelineComponent | None:
    return self._component_index.get(component_id)

manifest_clean

manifest_clean() -> Iterator[
    tuple[KubernetesManifest, ...]
]

Source code in kpops/pipeline.py

def manifest_clean(self) -> Iterator[tuple[KubernetesManifest, ...]]:
    for component in self.components:
        yield component.manifest_clean()

manifest_deploy

manifest_deploy() -> Iterator[
    tuple[KubernetesManifest, ...]
]

Source code in kpops/pipeline.py

def manifest_deploy(self) -> Iterator[tuple[KubernetesManifest, ...]]:
    for component in self.components:
        yield component.manifest_deploy()

manifest_destroy

manifest_destroy() -> Iterator[
    tuple[KubernetesManifest, ...]
]

Source code in kpops/pipeline.py

def manifest_destroy(self) -> Iterator[tuple[KubernetesManifest, ...]]:
    for component in self.components:
        yield component.manifest_destroy()

manifest_reset

manifest_reset() -> Iterator[
    tuple[KubernetesManifest, ...]
]

Source code in kpops/pipeline.py

def manifest_reset(self) -> Iterator[tuple[KubernetesManifest, ...]]:
    for component in self.components:
        yield component.manifest_reset()

remove

remove(component_id: str) -> None

Source code in kpops/pipeline.py

def remove(self, component_id: str) -> None:
    self._component_index.pop(component_id)

reset async

reset(dry_run: bool, parallel: bool = False) -> None

Reset pipeline steps.

PARAMETER	DESCRIPTION
dry_run	Whether to dry run the command or execute it. TYPE: bool
parallel	Enable or disable parallel execution of pipeline steps. TYPE: bool DEFAULT: False

Source code in kpops/pipeline.py

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
    )

to_yaml

to_yaml() -> str

Source code in kpops/pipeline.py

def to_yaml(self) -> str:
    return yaml.dump(self.generate(), sort_keys=False, Dumper=CustomSafeDumper)

validate

validate() -> None

Source code in kpops/pipeline.py

def validate(self) -> None:
    self.__validate_graph()