Repository: kedro-org/kedro
File: kedro/framework/hooks/manager.py (Python)

Code before:
from collections.abc import Iterable
from inspect import isclass
from typing import Any

from pluggy import PluginManager

from .markers import HOOK_NAMESPACE
from .specs import (
    DataCatalogSpecs,
    DatasetSpecs,
    KedroContextSpecs,
    NodeSpecs,
    PipelineSpecs,
)

_PLUGIN_HOOKS = "kedro.hooks"  # entry-point to load hooks from for installed plugins

logger = logging.getLogger(__name__)


def _create_hook_manager() -> PluginManager:
    """Create a new PluginManager instance and register Kedro's hook specs."""
    manager = PluginManager(HOOK_NAMESPACE)
    manager.trace.root.setwriter(
        logger.debug
        if logging.getLevelName(logger.getEffectiveLevel()) == "DEBUG"
        else None
    )
    manager.enable_tracing()
    manager.add_hookspecs(NodeSpecs)
    manager.add_hookspecs(PipelineSpecs)
    manager.add_hookspecs(DataCatalogSpecs)
    manager.add_hookspecs(DatasetSpecs)
    manager.add_hookspecs(KedroContextSpecs)
    return manager


def _register_hooks(hook_manager: PluginManager, hooks: Iterable[Any]) -> None:
    """Register all hooks as specified in ``hooks`` with the global ``hook_manager``.

    Args:
        hook_manager: Hook manager instance to register the hooks with.
        hooks: Hooks that need to be registered.

    """
    for hooks_collection in hooks:
        # Sometimes users might call hook registration more than once, in which
        # case hooks have already been registered, so we perform a simple check
        # here to avoid an error being raised and break user's workflow.
        if not hook_manager.is_registered(hooks_collection):
            if isclass(hooks_collection):

Review comment (bug):
```suggestion
        if logger.getEffectiveLevel() == logging.DEBUG
```

Code after:
from collections.abc import Iterable
from inspect import isclass
from typing import Any

from pluggy import PluginManager

from .markers import HOOK_NAMESPACE
from .specs import (
    DataCatalogSpecs,
    DatasetSpecs,
    KedroContextSpecs,
    NodeSpecs,
    PipelineSpecs,
)

_PLUGIN_HOOKS = "kedro.hooks"  # entry-point to load hooks from for installed plugins

logger = logging.getLogger(__name__)


def _create_hook_manager() -> PluginManager:
    """Create a new PluginManager instance and register Kedro's hook specs."""
    manager = PluginManager(HOOK_NAMESPACE)
    manager.trace.root.setwriter(
        logger.debug if logger.getEffectiveLevel() == logging.DEBUG else None
    )
    manager.enable_tracing()
    manager.add_hookspecs(NodeSpecs)
    manager.add_hookspecs(PipelineSpecs)
    manager.add_hookspecs(DataCatalogSpecs)
    manager.add_hookspecs(DatasetSpecs)
    manager.add_hookspecs(KedroContextSpecs)
    return manager


def _register_hooks(hook_manager: PluginManager, hooks: Iterable[Any]) -> None:
    """Register all hooks as specified in ``hooks`` with the global ``hook_manager``.

    Args:
        hook_manager: Hook manager instance to register the hooks with.
        hooks: Hooks that need to be registered.

    """
    for hooks_collection in hooks:
        # Sometimes users might call hook registration more than once, in which
        # case hooks have already been registered, so we perform a simple check
        # here to avoid an error being raised and break user's workflow.
        if not hook_manager.is_registered(hooks_collection):
            if isclass(hooks_collection):
                raise TypeError(
                    "KedroSession expects hooks to be registered as instances. "