Class BaseNode

scrapegraphai/nodes/base_node.py:12–235

An abstract base class for nodes in a graph-based workflow, designed to perform specific actions when executed. Attributes: node_name (str): The unique identifier name for the node. input (str): Boolean expression defining the input keys needed from the state. o

class BaseNode(ABC):
    """
    An abstract base class for nodes in a graph-based workflow,
    designed to perform specific actions when executed.

    Attributes:
        node_name (str): The unique identifier name for the node.
        input (str): Boolean expression defining the input keys needed from the state.
        output (List[str]): List of
        min_input_len (int): Minimum required number of input keys.
        node_config (Optional[dict]): Additional configuration for the node.
        logger (logging.Logger): The centralized root logger

    Args:
        node_name (str): Name for identifying the node.
        node_type (str): Type of the node; must be 'node' or 'conditional_node'.
        input (str): Expression defining the input keys needed from the state.
        output (List[str]): List of output keys to be updated in the state.
        min_input_len (int, optional): Minimum required number of input keys; defaults to 1.
        node_config (Optional[dict], optional): Additional configuration
                                                for the node; defaults to None.

    Raises:
        ValueError: If `node_type` is not one of the allowed types.

    Example:
        >>> class MyNode(BaseNode):
        ...     def execute(self, state):
        ...         # Implementation of node logic here
        ...         return state
        ...
        >>> my_node = MyNode("ExampleNode", "node", "input_spec", ["output_spec"])
        >>> updated_state = my_node.execute({'key': 'value'})
        {'key': 'value'}
    """

    def __init__(
        self,
        node_name: str,
        node_type: str,
        input: str,
        output: List[str],
        min_input_len: int = 1,
        node_config: Optional[dict] = None,
    ):
        self.node_name = node_name
        self.input = input
        self.output = output
        self.min_input_len = min_input_len
        self.node_config = node_config
        self.logger = get_logger()

        if node_type not in ["node", "conditional_node"]:
            raise ValueError(
                f"node_type must be 'node' or 'conditional_node', got '{node_type}'"
            )
        self.node_type = node_type