Repository: microsoft/promptflow
File: src/promptflow/promptflow/_sdk/entities/_flow.py (Python)

Code before:
        return kwargs.pop("async_call", False)

    @classmethod
    def load(
        cls,
        source: Union[str, PathLike],
        entry: str = None,
        raise_error=True,
        **kwargs,
    ):
        from promptflow._sdk.entities._eager_flow import EagerFlow

        source_path = Path(source)
        if not source_path.exists():
            raise UserErrorException(f"Source {source_path.absolute().as_posix()} does not exist")
        flow_path = resolve_flow_path(source_path)
        if not flow_path.exists():
            raise UserErrorException(f"Flow file {flow_path.absolute().as_posix()} does not exist")
        if flow_path.suffix in [".yaml", ".yml"]:
            # read flow file to get hash
            with open(flow_path, "r", encoding=DEFAULT_ENCODING) as f:
                flow_content = f.read()
                data = load_yaml_string(flow_content)
                content_hash = hash(flow_content)
            is_eager_flow = cls._is_eager_flow(data)
            is_async_flow = cls._is_async_flow(kwargs)
            if is_eager_flow:
                return EagerFlow._load(path=flow_path, data=data, raise_error=raise_error, **kwargs)
            else:
                # TODO: schema validation and warning on unknown fields
                if is_async_flow:
                    return AsyncProtectedFlow._load(path=flow_path, dag=data, content_hash=content_hash, **kwargs)
                else:
                    return ProtectedFlow._load(path=flow_path, dag=data, content_hash=content_hash, **kwargs)
        # if non-YAML file is provided, raise user error exception
        raise UserErrorException("Source must be a directory or a 'flow.dag.yaml' file")

    def _init_executable(self, tuning_node=None, variant=None):
        from promptflow._sdk._submitter import variant_overwrite_context
        from promptflow.contracts.flow import Flow as ExecutableFlow

        if not tuning_node and not variant:
            # for DAG flow, use data to init executable to improve performance
            return ExecutableFlow._from_dict(flow_dag=self._data, working_dir=self.code)

        # TODO: check if there is potential bug here
        # this is a little wired:
        # 1. the executable is created from a temp folder when there is additional includes
        # 2. after the executable is returned, the temp folder is deleted
        with variant_overwrite_context(self, tuning_node, variant) as flow:


Review comment (suggestion):
Directly check async_call param here instead using a function? Since async is now a capability of `Flow` object

Code after:
        source: Union[str, PathLike],
        entry: str = None,
        raise_error=True,
        **kwargs,
    ):
        from promptflow._sdk.entities._eager_flow import EagerFlow

        source_path = Path(source)
        if not source_path.exists():
            raise UserErrorException(f"Source {source_path.absolute().as_posix()} does not exist")
        flow_path = resolve_flow_path(source_path)
        if not flow_path.exists():
            raise UserErrorException(f"Flow file {flow_path.absolute().as_posix()} does not exist")
        if flow_path.suffix in [".yaml", ".yml"]:
            # read flow file to get hash
            with open(flow_path, "r", encoding=DEFAULT_ENCODING) as f:
                flow_content = f.read()
                data = load_yaml_string(flow_content)
                content_hash = hash(flow_content)
            is_eager_flow = cls._is_eager_flow(data)
            is_async_call = kwargs.pop("is_async_call", False)
            if is_eager_flow:
                return EagerFlow._load(path=flow_path, data=data, raise_error=raise_error, **kwargs)
            else:
                # TODO: schema validation and warning on unknown fields
                if is_async_call:
                    return AsyncProtectedFlow._load(path=flow_path, dag=data, content_hash=content_hash, **kwargs)
                else:
                    return ProtectedFlow._load(path=flow_path, dag=data, content_hash=content_hash, **kwargs)
        # if non-YAML file is provided, raise user error exception
        raise UserErrorException("Source must be a directory or a 'flow.dag.yaml' file")

    def _init_executable(self, tuning_node=None, variant=None):
        from promptflow._sdk._submitter import variant_overwrite_context
        from promptflow.contracts.flow import Flow as ExecutableFlow

        if not tuning_node and not variant:
            # for DAG flow, use data to init executable to improve performance
            return ExecutableFlow._from_dict(flow_dag=self._data, working_dir=self.code)

        # TODO: check if there is potential bug here
        # this is a little wired:
        # 1. the executable is created from a temp folder when there is additional includes
        # 2. after the executable is returned, the temp folder is deleted
        with variant_overwrite_context(self, tuning_node, variant) as flow:

            return ExecutableFlow.from_yaml(flow_file=flow.path, working_dir=flow.code)

    def __eq__(self, other):
        if isinstance(other, Flow):
            return self._content_hash == other._content_hash and self.context == other.context