    def acquire_op(self, key, hub_op_id: str, op_args: List, op_kws: Dict[str, any], tag: str, latest: bool) -> Operator:
        """
        Instruct the `OperatorPool` to reserve and return the
        specified operator for use in the executor.

        Args:
            key: (`str`)
            hub_op_id: (`str`)
            op_args: (`List`)
                Operator init parameters with args
            op_kws: (`Dict[str, any]`)
                Operator init parameters with kwargs
            tag: (`str`)
                The tag of operator
            latest (`bool`):
                Whether to download the latest files.

        Returns:
            (`towhee.operator.Operator`)
                The operator instance reserved for the caller.
        """

        # Load the operator if the computed key does not exist in the operator
        # dictionary.
        with self._lock:
            storage = self._all_ops.get(key, None)
            if storage is None:
                storage = _OperatorStorage()
                self._all_ops[key] = storage

            if not storage.op_available():
                op = self._op_loader.load_operator(hub_op_id, op_args, op_kws, tag, latest)
                storage.put(op, True)
                op.key = key
            return storage.get()
Instruct the `OperatorPool` to reserve and return the
specified operator for use in the executor.

Args:
key: (`str`)
hub_op_id: (`str`)
op_args: (`List`)
Operator init parameters with args
op_kws: (`Dict[str, any]`)
Operator init parameters with kwargs
tag: (`str`)
The tag of operator
latest (`bool`):
Whether to download the latest files.

Returns:
(`towhee.operator.Operator`)
The operator instance reserved for the caller.
towhee/runtime/operator_manager/operator_pool.py