Repository: microsoft/promptflow
File: src/promptflow-devkit/promptflow/_proxy/_python_executor_proxy.py (Python)

Code before:
        **kwargs,
    ) -> "PythonExecutorProxy":
        # Check if the method accepts kwargs in case of customer using an outdated version of core package.
        if can_accept_kwargs(FlowExecutor.create):
            flow_executor = FlowExecutor.create(
                flow_file, connections, working_dir, storage=storage, raise_ex=False, init_kwargs=init_kwargs, **kwargs
            )
        else:
            flow_executor = FlowExecutor.create(
                flow_file, connections, working_dir, storage=storage, raise_ex=False, init_kwargs=init_kwargs
            )
        return cls(flow_executor)

    @property
    def has_aggregation(self) -> bool:
        return self._flow_executor.has_aggregation_node

    async def exec_aggregation_async(
        self,
        batch_inputs: Mapping[str, Any],
        aggregation_inputs: Mapping[str, Any],
        run_id: Optional[str] = None,
    ) -> AggregationResult:
        if isinstance(self._flow_executor, ScriptExecutor):
            return self._flow_executor._exec_aggregation(aggregation_inputs, run_id=run_id)
        return self._flow_executor.exec_aggregation(batch_inputs, aggregation_inputs, run_id=run_id)

    async def _exec_batch(
        self,
        batch_inputs: List[Mapping[str, Any]],
        output_dir: Path,
        run_id: Optional[str] = None,
        batch_timeout_sec: Optional[int] = None,
        line_timeout_sec: Optional[int] = None,
        worker_count: Optional[int] = None,
    ) -> Tuple[List[LineResult], bool]:
        # TODO: Refine the logic here since the script executor actually doesn't have the 'node' concept
        if isinstance(self._flow_executor, ScriptExecutor):
            run_tracker = RunTracker(self._flow_executor._storage)
        else:
            run_tracker = self._flow_executor._run_tracker

        with run_tracker.node_log_manager:
            OperationContext.get_instance().run_mode = RunMode.Batch.name
            if self._flow_executor._flow_file is None:
                raise UnexpectedError(
                    "Unexpected error occurred while init FlowExecutor. Error details: flow file is missing."
                )

            if batch_timeout_sec:
                bulk_logger.info(f"The timeout for the batch run is {batch_timeout_sec} seconds.")

Review comment (performance):
It is not a good idea to have `if else` here, I think we could implement`exec_aggregation` with the same interface, just ignore batch_inputs in the imlementation.

When we inherit the class, we are going to implement the same interface to avoid such if/else

Code after:
        **kwargs,
    ) -> "PythonExecutorProxy":
        # Check if the method accepts kwargs in case of customer using an outdated version of core package.
        if can_accept_kwargs(FlowExecutor.create):
            flow_executor = FlowExecutor.create(
                flow_file, connections, working_dir, storage=storage, raise_ex=False, init_kwargs=init_kwargs, **kwargs
            )
        else:
            flow_executor = FlowExecutor.create(
                flow_file, connections, working_dir, storage=storage, raise_ex=False, init_kwargs=init_kwargs
            )
        return cls(flow_executor)

    @property
    def has_aggregation(self) -> bool:
        return self._flow_executor.has_aggregation_node

    async def exec_aggregation_async(
        self,
        batch_inputs: Mapping[str, Any],
        aggregation_inputs: Mapping[str, Any],
        run_id: Optional[str] = None,
    ) -> AggregationResult:
        return self._flow_executor.exec_aggregation(batch_inputs, aggregation_inputs, run_id=run_id)

    async def _exec_batch(
        self,
        batch_inputs: List[Mapping[str, Any]],
        output_dir: Path,
        run_id: Optional[str] = None,
        batch_timeout_sec: Optional[int] = None,
        line_timeout_sec: Optional[int] = None,
        worker_count: Optional[int] = None,
    ) -> Tuple[List[LineResult], bool]:
        # TODO: Refine the logic here since the script executor actually doesn't have the 'node' concept
        if isinstance(self._flow_executor, ScriptExecutor):
            run_tracker = RunTracker(self._flow_executor._storage)
        else:
            run_tracker = self._flow_executor._run_tracker

        with run_tracker.node_log_manager:
            OperationContext.get_instance().run_mode = RunMode.Batch.name
            if self._flow_executor._flow_file is None:
                raise UnexpectedError(
                    "Unexpected error occurred while init FlowExecutor. Error details: flow file is missing."
                )

            if batch_timeout_sec:
                bulk_logger.info(f"The timeout for the batch run is {batch_timeout_sec} seconds.")

            with LineExecutionProcessPool(