Upload operators.py with huggingface_hub

operators.py

@@ -41,6 +41,7 @@ from abc import abstractmethod
 from collections import Counter
 from copy import deepcopy
 from dataclasses import field
+from importlib import import_module
 from itertools import zip_longest
 from random import Random
 from typing import (
@@ -58,7 +59,7 @@ from typing import (
 import requests
 
 from .artifact import Artifact, fetch_artifact
-from .dataclass import NonPositionalField
+from .dataclass import NonPositionalField, OptionalField
 from .dict_utils import dict_delete, dict_get, dict_set, is_subpath
 from .operator import (
     MultiStream,
@@ -74,11 +75,14 @@ from .operator import (
     StreamInstanceOperator,
 )
 from .random_utils import new_random_generator
+from .settings_utils import get_settings
 from .stream import Stream
 from .text_utils import nested_tuple_to_string
 from .type_utils import isoftype
 from .utils import flatten_dict
 
+settings = get_settings()
+
 
 class FromIterables(StreamInitializerOperator):
     """Creates a MultiStream from a dict of named iterables.
@@ -484,7 +488,7 @@ class AddConstant(FieldOperator):
 
 
 class Augmentor(StreamInstanceOperator):
-    """A stream that augments the values of either the task input fields before rendering with the template,  or the  input passed to the model after rendering of the template.
+    """A stream operator that augments the values of either the task input fields before rendering with the template,  or the input passed to the model after rendering of the template.
 
     Args:
         augment_model_input: Whether to augment the input to the model.
@@ -564,9 +568,9 @@ class NullAugmentor(Augmentor):
 
 
 class AugmentWhitespace(Augmentor):
-    """Augments the inputs by replace existing whitespace with other whitespace.
+    """Augments the inputs by replacing existing whitespaces with other whitespaces.
 
-    Currently each whitespace is replaced by a random choice of 1-3 whitespace charaters (spcae, tab, newline).
+    Currently, each whitespace is replaced by a random choice of 1-3 whitespace characters (space, tab, newline).
     """
 
     def process_value(self, value: Any) -> Any:
@@ -1094,7 +1098,7 @@ class ArtifactFetcherMixin:
         return cls.cache[artifact_identifier]
 
 
-class ApplyOperatorsField(StreamInstanceOperator, ArtifactFetcherMixin):
+class ApplyOperatorsField(StreamInstanceOperator):
     """Applies value operators to each instance in a stream based on specified fields.
 
     Args:
@@ -1215,30 +1219,60 @@ class FilterByCondition(SingleStreamOperator):
         return True
 
 
-class FilterByQuery(SingleStreamOperator):
+class ComputeExpressionMixin(Artifact):
+    """Computes an expression expressed over fields of an instance.
+
+    Args:
+        expression (str): the expression, in terms of names of fields of an instance
+        imports_list (List[str]): list of names of imports needed for the evaluation of the expression
+    """
+
+    expression: str
+    imports_list: List[str] = OptionalField(default_factory=list)
+
+    def prepare(self):
+        # can not do the imports here, because object does not pickle with imports
+        self.globs = {}
+        self.to_import = True
+
+    def compute_expression(self, instance: dict) -> Any:
+        if self.to_import:
+            for module_name in self.imports_list:
+                self.globs[module_name] = import_module(module_name)
+            self.to_import = False
+
+        if settings.allow_unverified_code:
+            return eval(self.expression, self.globs, instance)
+
+        raise ValueError(
+            f"Cannot run expression by {self} when unitxt.settings.allow_unverified_code=False either set it to True or set {settings.allow_unverified_code_key} environment variable."
+        )
+
+
+class FilterByExpression(SingleStreamOperator, ComputeExpressionMixin):
     """Filters a stream, yielding only instances which fulfil a condition specified as a string to be python's eval-uated.
 
     Raises an error if a field participating in the specified condition is missing from the instance
 
     Args:
-       query (str): a condition over fields of the instance, to be processed by python's eval()
+       expression (str): a condition over fields of the instance, to be processed by python's eval()
+       imports_list (List[str]): names of imports needed for the eval of the query (e.g. 're', 'json')
        error_on_filtered_all (bool, optional): If True, raises an error if all instances are filtered out. Defaults to True.
 
     Examples:
-       FilterByQuery(query = "a > 4") will yield only instances where "a">4
-       FilterByQuery(query = "a <= 4 and b > 5") will yield only instances where the value of field "a" is not exceeding 4 and in field "b" -- greater than 5
-       FilterByQuery(query = "a in [4, 8]") will yield only instances where "a" is 4 or 8
-       FilterByQuery(query = "a not in [4, 8]") will yield only instances where "a" is neither 4 nor 8
+       FilterByExpression(expression = "a > 4") will yield only instances where "a">4
+       FilterByExpression(expression = "a <= 4 and b > 5") will yield only instances where the value of field "a" is not exceeding 4 and in field "b" -- greater than 5
+       FilterByExpression(expression = "a in [4, 8]") will yield only instances where "a" is 4 or 8
+       FilterByExpression(expression = "a not in [4, 8]") will yield only instances where "a" is neither 4 nor 8
 
     """
 
-    query: str
     error_on_filtered_all: bool = True
 
     def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
         yielded = False
         for instance in stream:
-            if eval(self.query, None, instance):
+            if self.compute_expression(instance):
                 yielded = True
                 yield instance
 
@@ -1248,33 +1282,33 @@ class FilterByQuery(SingleStreamOperator):
             )
 
 
-class ExecuteQuery(StreamInstanceOperator):
-    """Compute an expression (query), expressed as a string to be eval-uated, over the instance's fields, and store the result in field to_field.
+class ExecuteExpression(StreamInstanceOperator, ComputeExpressionMixin):
+    """Compute an expression, specified as a string to be eval-uated, over the instance's fields, and store the result in field to_field.
 
     Raises an error if a field mentioned in the query is missing from the instance.
 
     Args:
-       query (str): an expression to be evaluated over the fields of the instance
+       expression (str): an expression to be evaluated over the fields of the instance
        to_field (str): the field where the result is to be stored into
+       imports_list (List[str]): names of imports needed for the eval of the query (e.g. 're', 'json')
 
     Examples:
        When instance {"a": 2, "b": 3} is process-ed by operator
-       ExecuteQuery(query="a+b", to_field = "c")
+       ExecuteExpression(expression="a+b", to_field = "c")
        the result is {"a": 2, "b": 3, "c": 5}
 
        When instance {"a": "hello", "b": "world"} is process-ed by operator
-       ExecuteQuery(query = "a+' '+b", to_field = "c")
+       ExecuteExpression(expression = "a+' '+b", to_field = "c")
        the result is {"a": "hello", "b": "world", "c": "hello world"}
 
     """
 
-    query: str
     to_field: str
 
     def process(
         self, instance: Dict[str, Any], stream_name: Optional[str] = None
     ) -> Dict[str, Any]:
-        instance[self.to_field] = eval(self.query, None, instance)
+        instance[self.to_field] = self.compute_expression(instance)
         return instance