Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/executor/manager.py (Python)

Code before:
    properties = input_schema.get("properties", {})

    for field_name, field_schema in properties.items():
        if "depends_on" in field_schema:
            dependencies = field_schema["depends_on"]

            # Check if dependent field has value
            has_value = data.get(field_name) is not None and data.get(field_name) != "" or (field_schema.get("default") is not None and field_schema.get("default") != "")
            must_have_value = field_name in required_fields

            # Check for missing dependencies when depenedent field is present
            missing_deps = [
                dep for dep in dependencies
                if not data.get(dep) or str(data[dep]).strip() == ""
            ]

            if (has_value or must_have_value) and missing_deps:
                return None, f"{error_prefix} Field {field_name} requires {', '.join(missing_deps)} to be set"

            # Check if field is required when dependencies are present
            has_all_deps = all(
                data.get(dep) and str(data[dep]).strip() != ""
                for dep in dependencies
            )
            if has_all_deps and not has_value:
                return None, f"{error_prefix} {field_name} is required when {', '.join(dependencies)} are set"


    # Last validation: Validate the input values against the schema.
    if error := json.validate_with_jsonschema(schema=input_schema, data=data):
        error_message = f"{error_prefix} {error}"
        logger.error(error_message)
        return None, error_message

    return data, node_block.name


class Executor:
    """
    This class contains event handlers for the process pool executor events.

    The main events are:
        on_node_executor_start: Initialize the process that executes the node.
        on_node_execution: Execution logic for a node.

        on_graph_executor_start: Initialize the process that executes the graph.
        on_graph_execution: Execution logic for a graph.

    The execution flow:
        1. Graph execution request is added to the queue.
        2. Graph executor loop picks the request from the queue.

Review comment (refactor):
Could you move this to `GraphModel.validate_graph`?

Code after:
    if error := json.validate_with_jsonschema(schema=input_schema, data=data):
        error_message = f"{error_prefix} {error}"
        logger.error(error_message)
        return None, error_message

    return data, node_block.name


class Executor:
    """
    This class contains event handlers for the process pool executor events.

    The main events are:
        on_node_executor_start: Initialize the process that executes the node.
        on_node_execution: Execution logic for a node.

        on_graph_executor_start: Initialize the process that executes the graph.
        on_graph_execution: Execution logic for a graph.

    The execution flow:
        1. Graph execution request is added to the queue.
        2. Graph executor loop picks the request from the queue.
        3. Graph executor loop submits the graph execution request to the executor pool.
      [on_graph_execution]
        4. Graph executor initialize the node execution queue.
        5. Graph executor adds the starting nodes to the node execution queue.
        6. Graph executor waits for all nodes to be executed.
      [on_node_execution]
        7. Node executor picks the node execution request from the queue.
        8. Node executor executes the node.
        9. Node executor enqueues the next executed nodes to the node execution queue.
    """

    @classmethod
    def on_node_executor_start(cls):
        configure_logging()
        set_service_name("NodeExecutor")
        redis.connect()
        cls.pid = os.getpid()
        cls.db_client = get_db_client()
        cls.creds_manager = IntegrationCredentialsManager()

        # Set up shutdown handlers
        cls.shutdown_lock = threading.Lock()
        atexit.register(cls.on_node_executor_stop)  # handle regular shutdown
        signal.signal(  # handle termination
            signal.SIGTERM, lambda _, __: cls.on_node_executor_sigterm()
        )

    @classmethod
    def on_node_executor_stop(cls):