ronantakizawa/codereview-bench huggingface dataset:
Repository: kedro-org/kedro
File: kedro/runner/runner.py (Python)

Code before:
"""``AbstractRunner`` is the base class for all ``Pipeline`` runner
implementations.
"""

from __future__ import annotations

import inspect
import logging
import os
import sys
import warnings
from abc import ABC, abstractmethod
from collections import Counter, deque
from concurrent.futures import (
    FIRST_COMPLETED,
    ProcessPoolExecutor,
    ThreadPoolExecutor,
    wait,
)
from itertools import chain
from typing import TYPE_CHECKING, Any

from pluggy import PluginManager

from kedro import KedroDeprecationWarning
from kedro.framework.hooks.manager import _NullPluginManager
from kedro.io import CatalogProtocol, MemoryDataset, SharedMemoryDataset
from kedro.pipeline import Pipeline
from kedro.runner.task import Task

# see https://github.com/python/cpython/blob/master/Lib/concurrent/futures/process.py#L114
_MAX_WINDOWS_WORKERS = 61

if TYPE_CHECKING:
    from collections.abc import Collection, Iterable

    from pluggy import PluginManager

    from kedro.pipeline.node import Node


class AbstractRunner(ABC):
    """``AbstractRunner`` is the base class for all ``Pipeline`` runner

Review comment (suggestion):
Would this import `multiprocessing` as a dependencies? I recalled in the past we have issues with `ShelveStore` because even importing the library cause issues on restricted environment like AWS Lambda.

Code after:
"""``AbstractRunner`` is the base class for all ``Pipeline`` runner
implementations.
"""

from __future__ import annotations

import inspect
import logging
import os
import sys
import warnings
from abc import ABC, abstractmethod
from collections import Counter, deque
from concurrent.futures import FIRST_COMPLETED, Executor, ProcessPoolExecutor, wait
from itertools import chain
from typing import TYPE_CHECKING, Any

from pluggy import PluginManager

from kedro import KedroDeprecationWarning
from kedro.framework.hooks.manager import _NullPluginManager
from kedro.io import CatalogProtocol, MemoryDataset, SharedMemoryDataset
from kedro.pipeline import Pipeline
from kedro.runner.task import Task

# see https://github.com/python/cpython/blob/master/Lib/concurrent/futures/process.py#L114
_MAX_WINDOWS_WORKERS = 61

if TYPE_CHECKING:
    from collections.abc import Collection, Iterable

    from pluggy import PluginManager

    from kedro.pipeline.node import Node


class AbstractRunner(ABC):
    """``AbstractRunner`` is the base class for all ``Pipeline`` runner
    implementations.
    """

    def __init__(
        self,
Repository: kedro-org/kedro
File: kedro/templates/project/{{ cookiecutter.repo_name }}/tests/test_run.py (Python)

Code before:
"""
This module contains example tests for a Kedro project.
Tests should be placed in ``src/tests``, in modules that mirror your
project's structure, and in files named test_*.py.
"""
from pathlib import Path

from kedro.framework.session import KedroSession
from kedro.framework.startup import bootstrap_project

# The tests below are here for the demonstration purpose
# and should be replaced with the ones testing the project
# functionality

class TestKedroRun:
    def test_kedro_run(self):
        bootstrap_project(Path.cwd())

        with KedroSession.create(project_path=Path.cwd()) as session:
            assert session.run() is not None


Review comment (suggestion):
I think it would be helpful for users to explain what's happening here. Adding a link to the docs in the comments could provide more clarity:
[Testing a Kedro Project](https://docs.kedro.org/en/stable/tutorial/test_a_project.html).

It might also be a good idea to update the docs to include a more detailed explanation of this testing example.

Code after:
"""
This module contains example tests for a Kedro project.
Tests should be placed in ``src/tests``, in modules that mirror your
project's structure, and in files named test_*.py.
"""
import pytest
from pathlib import Path
from kedro.framework.session import KedroSession
from kedro.framework.startup import bootstrap_project

# The tests below are here for the demonstration purpose
# and should be replaced with the ones testing the project
# functionality

class TestKedroRun:
    def test_kedro_run_no_pipeline(self):
    # This example test expects a pipeline run failure, since
    # the default project template contains no pipelines.
        bootstrap_project(Path.cwd())

        with pytest.raises(Exception) as excinfo:
            with KedroSession.create(project_path=Path.cwd()) as session:
                session.run()

        assert "Pipeline contains no nodes" in str(excinfo.value)
Repository: kedro-org/kedro
File: kedro/__init__.py (Python)

Code before:
__version__ = "0.19.10"


class KedroDeprecationWarning(DeprecationWarning):
    """Custom class for warnings about deprecated Kedro features."""


class KedroPythonVersionWarning(UserWarning):
    """Custom class for warnings about incompatibilities with Python versions."""


if not sys.warnoptions:
    warnings.simplefilter("default", KedroDeprecationWarning)
    warnings.simplefilter("error", KedroPythonVersionWarning)

if sys.version_info >= (3, 14):
    warnings.warn(
        """Kedro is not yet fully compatible with this Python version.
To proceed at your own risk and ignore this warning,
run Kedro with `python -W "default:Kedro is not yet fully compatible" -m kedro ...`
or set the PYTHONWARNINGS environment variable accordingly.""",
        KedroPythonVersionWarning,
    )


def load_ipython_extension(ipython: Any) -> None:
    import kedro.ipython

    kedro.ipython.load_ipython_extension(ipython)


Review comment (suggestion):
Maybe

```python
from __future__ import annotations

import typing

if typing.TYPE_CHECKING:
    from IPython.core.interactiveshell import InteractiveShellABC

...

def load_ipython_extension(ipython: InteractiveShellABC) -> None:
    ...
```

? https://docs.python.org/3/library/typing.html#typing.TYPE_CHECKING

Code after:
if TYPE_CHECKING:
    from IPython.core.interactiveshell import InteractiveShell

import sys
import warnings

__version__ = "0.19.10"


class KedroDeprecationWarning(DeprecationWarning):
    """Custom class for warnings about deprecated Kedro features."""


class KedroPythonVersionWarning(UserWarning):
    """Custom class for warnings about incompatibilities with Python versions."""


if not sys.warnoptions:
    warnings.simplefilter("default", KedroDeprecationWarning)
    warnings.simplefilter("error", KedroPythonVersionWarning)

if sys.version_info >= (3, 14):
    warnings.warn(
        """Kedro is not yet fully compatible with this Python version.
To proceed at your own risk and ignore this warning,
run Kedro with `python -W "default:Kedro is not yet fully compatible" -m kedro ...`
or set the PYTHONWARNINGS environment variable accordingly.""",
        KedroPythonVersionWarning,
    )


def load_ipython_extension(ipython: InteractiveShell) -> None:
    import kedro.ipython

    kedro.ipython.load_ipython_extension(ipython)
Repository: kedro-org/kedro
File: kedro/io/kedro_data_catalog.py (Python)

Code before:
            Copy of the current object.
        """
        if extra_dataset_patterns:
            self._config_resolver.add_runtime_patterns(extra_dataset_patterns)
        return self

    def exists(self, name: str) -> bool:
        """Checks whether registered dataset exists by calling its `exists()`
        method. Raises a warning and returns False if `exists()` is not
        implemented.

        Args:
            name: A dataset to be checked.

        Returns:
            Whether the dataset output exists.

        """
        try:
            dataset = self._get_dataset(name)
        except DatasetNotFoundError:
            return False
        return dataset.exists()


def _compile_pattern(regex: str, regex_flags: int | re.RegexFlag) -> re.Pattern:
    try:
        pattern = re.compile(regex, flags=regex_flags)
    except re.error as exc:
        raise SyntaxError(f"Invalid regular expression provided: '{regex}'") from exc

    return pattern


Review comment (suggestion):
Maybe call this `_compile_regex_pattern` to not confuse this with the patterns for dataset factories. 

Code after:
        # This method was simplified to add memory datasets only, since
        # adding AbstractDataset can be done via add() method
        for ds_name, ds_data in feed_dict.items():
            self.add(ds_name, MemoryDataset(data=ds_data), replace)  # type: ignore[abstract]

    def shallow_copy(
        self, extra_dataset_patterns: Patterns | None = None
    ) -> KedroDataCatalog:
        # TODO: remove when removing old catalog
        """Returns a shallow copy of the current object.

        Returns:
            Copy of the current object.
        """
        if extra_dataset_patterns:
            self._config_resolver.add_runtime_patterns(extra_dataset_patterns)
        return self

    def exists(self, name: str) -> bool:
        """Checks whether registered dataset exists by calling its `exists()`
        method. Raises a warning and returns False if `exists()` is not
        implemented.

        Args:
            name: A dataset to be checked.

        Returns:
            Whether the dataset output exists.

        """
        try:
            dataset = self._get_dataset(name)
        except DatasetNotFoundError:
            return False
        return dataset.exists()
Repository: kedro-org/kedro
File: tests/framework/project/test_logging.py (Python)

Code before:
    assert expected_message in "".join(caplog.messages).strip("\n")


def test_logger_without_rich_markup():
    class CustomHandler(logging.Handler):
        def __init__(self):
            super().__init__()
            self.records = []

        def emit(self, record):
            self.records.append(record)

    data = ("dummy",)
    catalog = DataCatalog.from_config({"dummy": {"type": "MemoryDataset"}})

    # Add a custom handler
    custom_handler = CustomHandler()
    root_logger = logging.getLogger()
    root_logger.addHandler(custom_handler)

    # Emit some logs
    assert not custom_handler.records
    catalog.save("dummy", data)
    assert custom_handler.records
    for record in custom_handler.records:
        assert "[dark_orange]" not in record.message


Review comment (suggestion):
This should sufficiently show that even with a dummy handler, it still capture the rich markup.
@astrojuanlu 

Code after:
    expected_message = "You can change this by setting the KEDRO_LOGGING_CONFIG environment variable accordingly."
    assert expected_message in "".join(caplog.messages).strip("\n")


def test_logger_without_rich_markup():
    class CustomHandler(logging.Handler):
        def __init__(self):
            super().__init__()
            self.records = []

        def emit(self, record):
            self.records.append(record)

    data = ("dummy",)
    catalog = DataCatalog.from_config({"dummy": {"type": "MemoryDataset"}})

    # Add a custom handler
    custom_handler = CustomHandler()
    root_logger = logging.getLogger()
    root_logger.addHandler(custom_handler)

    # Emit some logs
    assert not custom_handler.records
    catalog.save("dummy", data)
    assert custom_handler.records

    for record in custom_handler.records:
        assert "[dark_orange]" not in record.message


def test_logger_with_invalid_markup_args():
    root_logger = logging.getLogger()
    rich_handler = RichHandler()
    root_logger.addHandler(rich_handler)
    var = "dummy"
    with pytest.raises(TypeError):
        root_logger.warn("dummy %s", var, extra={"rich_format": "invalid_arg"})
Repository: kedro-org/kedro
File: features/environment.py (Python)

Code before:
        context.kedro_install_venv_dir = kedro_install_venv_dir
        context = _setup_context_with_venv(context, kedro_install_venv_dir)

        call(
            [
                context.python,
                "-m",
                "pip",
                "install",
                "-U",
                "pip>=21.2",
            ],
            env=context.env,
        )
        call(
            [
                context.python,
                "-m",
                "pip",
                "install",
                "-e",
                ".",
                "toml>=0.10.0",  # TODO(deepyaman): Migrate `kedro-telemetry` off `toml`
            ],
            env=context.env,
        )
        return context


def _install_project_requirements(context):
    install_reqs = (
        Path("kedro/templates/project/{{ cookiecutter.repo_name }}/requirements.txt")
        .read_text(encoding="utf-8")
        .splitlines()
    )
    install_reqs = [req for req in install_reqs if "{" not in req and "#" not in req]
    install_reqs.append("kedro-datasets[pandas-csvdataset]")
    call([context.pip, "install", *install_reqs], env=context.env)
    return context


Review comment (suggestion):
```suggestion
        call([context.python, "-m", "pip", "install", "-e", "."], env=context.env)
```

Code after:
        context.kedro_install_venv_dir = kedro_install_venv_dir
        context = _setup_context_with_venv(context, kedro_install_venv_dir)

        call(
            [
                context.python,
                "-m",
                "pip",
                "install",
                "-U",
                "pip>=21.2",
            ],
            env=context.env,
        )
        call([context.python, "-m", "pip", "install", "-e", "."], env=context.env)
        return context


def _install_project_requirements(context):
    install_reqs = (
        Path("kedro/templates/project/{{ cookiecutter.repo_name }}/requirements.txt")
        .read_text(encoding="utf-8")
        .splitlines()
    )
    install_reqs = [req for req in install_reqs if "{" not in req and "#" not in req]
    install_reqs.append("kedro-datasets[pandas-csvdataset]")
    call([context.pip, "install", *install_reqs], env=context.env)
    return context
Repository: kedro-org/kedro
File: kedro/framework/cli/project.py (Python)

Code before:
    is_flag=True,
    help=ONLY_MISSING_OUTPUTS_HELP,
)
def run(  # noqa: PLR0913
    tags: str,
    env: str,
    runner: str,
    is_async: bool,
    node_names: str,
    to_nodes: str,
    from_nodes: str,
    from_inputs: str,
    to_outputs: str,
    load_versions: dict[str, str] | None,
    pipeline: str,
    pipelines: list[str],
    config: str,
    conf_source: str,
    params: dict[str, Any],
    namespaces: str,
    only_missing_outputs: bool,
) -> dict[str, Any]:
    """Run the pipeline."""

    if pipeline and pipelines:
        raise click.UsageError(
            "Options '--pipeline' and '--pipelines' cannot be used together"
        )

    pipelines_to_run = list(set(pipelines)) if pipelines else None

    runner_obj = load_obj(runner or "SequentialRunner", "kedro.runner")
    tuple_tags = tuple(tags)
    tuple_node_names = tuple(node_names)

    with KedroSession.create(
        env=env, conf_source=conf_source, runtime_params=params
    ) as session:
        return session.run(
            tags=tuple_tags,
            runner=runner_obj(is_async=is_async),
            node_names=tuple_node_names,
            from_nodes=from_nodes,
            to_nodes=to_nodes,
            from_inputs=from_inputs,
            to_outputs=to_outputs,
            load_versions=load_versions,
            pipeline_name=pipeline,
            pipeline_names=pipelines_to_run,
            namespaces=namespaces,
            only_missing_outputs=only_missing_outputs,

Review comment (suggestion):
Can we raise `KedroCliError` here instead of `click.UsageError`

Code after:
    "--only-missing-outputs",
    is_flag=True,
    help=ONLY_MISSING_OUTPUTS_HELP,
)
def run(  # noqa: PLR0913
    tags: str,
    env: str,
    runner: str,
    is_async: bool,
    node_names: str,
    to_nodes: str,
    from_nodes: str,
    from_inputs: str,
    to_outputs: str,
    load_versions: dict[str, str] | None,
    pipeline: str,
    pipelines: list[str],
    config: str,
    conf_source: str,
    params: dict[str, Any],
    namespaces: str,
    only_missing_outputs: bool,
) -> dict[str, Any]:
    """Run the pipeline."""

    if pipeline and pipelines:
        raise KedroCliError(
            "Options '--pipeline' and '--pipelines' cannot be used together"
        )

    pipelines_to_run = list(set(pipelines)) if pipelines else None

    runner_obj = load_obj(runner or "SequentialRunner", "kedro.runner")
    tuple_tags = tuple(tags)
    tuple_node_names = tuple(node_names)

    with KedroSession.create(
        env=env, conf_source=conf_source, runtime_params=params
    ) as session:
        return session.run(
            tags=tuple_tags,
            runner=runner_obj(is_async=is_async),
            node_names=tuple_node_names,
            from_nodes=from_nodes,
            to_nodes=to_nodes,
            from_inputs=from_inputs,
            to_outputs=to_outputs,
            load_versions=load_versions,
            pipeline_name=pipeline,
            pipeline_names=pipelines_to_run,
            namespaces=namespaces,
Repository: kestra-io/kestra
File: core/src/main/java/io/kestra/core/runners/RunContext.java (Java)

Code before:

    /**
     * @deprecated use #namespaceKv(String) instead
     */
    @Deprecated(since = "1.1.0", forRemoval = true)
    public StateStore stateStore() {
        return new StateStore(this, true);
    }

    /**
     * Get access to local paths of the host machine.
     */
    public abstract LocalPath localPath();

    public record FlowInfo(String tenantId, String namespace, String id, Integer revision) {
    }

    public abstract boolean isInitialized();

    /**
     * Finds a flow for a taskrun.
     * This method checks that the flow related to the taskrun is allowed to access the target flow (via ACL check)
     *
     * @see #acl() which is used to check the taskrun ACL to the target flow.
     */
    public abstract FlowInterface findFlow(String tenantId, String namespace, String flowId, Optional<Integer> revision, TaskRun taskRun);

    /**
     * Get access to the ACL checker.
     * Plugins are responsible for using the ACL checker when they access restricted resources, so Namespace ACLs are used (EE).
     */
    public abstract AclChecker acl();
}


Review comment (suggestion):
I would prefer to have a facade class (e.g., `FlowAccessor` ) here to access flow like this: `runContext.flows.find()`, following the DSL style we already have with other resources (i.e, storage, workingDir, etc). Agree, that for the moment we only have a single method. But maybe if need to add one later this will have no impact on the RunContext class.

Code after:
     * @deprecated use #namespaceKv(String) instead
     */
    @Deprecated(since = "1.1.0", forRemoval = true)
    public StateStore stateStore() {
        return new StateStore(this, true);
    }

    /**
     * Get access to local paths of the host machine.
     */
    public abstract LocalPath localPath();

    public record FlowInfo(String tenantId, String namespace, String id, Integer revision) {
    }

    /**
     * @deprecated there is no legitimate use case of this method outside the run context internal self-usage, so it should not be part of the interface
     */
    @Deprecated(since = "1.2.0", forRemoval = true)
    public abstract boolean isInitialized();

    /**
     * Get access to the ACL checker.
     * Plugins are responsible for using the ACL checker when they access restricted resources, for example,
     * when Namespace ACLs are used (EE).
     */
    public abstract AclChecker acl();
}
Repository: kestra-io/kestra
File: core/src/main/java/io/kestra/core/runners/AclChecker.java (Java)

Code before:
package io.kestra.core.runners;

import io.kestra.core.services.FlowService;

import java.util.List;
import java.util.Objects;

public interface AclChecker {

    void checkAllowAllNamespace();

    void checkAllowNamespace(String namespace);

    void checkAllowNamespaces(List<String> namespaces);

    void checkAllowResources(AllowedResources allowedResources);

    interface AllowedResources {
        void checkAllowed(RunContext.FlowInfo flowInfo);
    }

    class AllowAllNamespace implements AllowedResources {
        private final FlowService flowService;

        AllowAllNamespace(FlowService flowService) {
            this.flowService = Objects.requireNonNull(flowService);
        }

        @Override
        public void checkAllowed(RunContext.FlowInfo flowInfo) {
            this.flowService.checkAllowedAllNamespaces(flowInfo.tenantId(), flowInfo.tenantId(), flowInfo.namespace());
        }
    }

    class AllowNamespaces implements AllowedResources {

Review comment (suggestion):
I think, we should create a dedicated un-checked exception for thoses methods, e.g., `ResourceAccessDeniedException` that will be explicilty documented in this interface.

Code after:
package io.kestra.core.runners;

import java.util.List;

/**
 * Check if the current taskrun has access to the requested resources.
 *
 * <p>
 * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
 *
 * @see AllowedResources
 */
public interface AclChecker {

    /**Tasks that need to access resources outside their namespace should use this interface to check ACL (Allowed namespaces in EE).
     * Allow all namespaces.
     * <p>
     * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
     */
    AllowedResources allowAllNamespaces();

    /**
     * Allow only the given namespace.
     * <p>
     * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
     */
    AllowedResources allowNamespace(String namespace);

    /**
     * Allow only the given namespaces.
     * <p>
     * IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
     */
    AllowedResources allowNamespaces(List<String> namespaces);