derrickaw commented on code in PR #38473:
URL: https://github.com/apache/beam/pull/38473#discussion_r3242033123
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -178,18 +178,38 @@ def _check_mapping_arguments(
raise ValueError(f'{transform_name} cannot specify "name" without "path"')
-# js2py's JsObjectWrapper object has a self-referencing __dict__ property
-# that cannot be pickled without implementing the __getstate__ and
-# __setstate__ methods.
-class _CustomJsObjectWrapper(JsObjectWrapper):
- def __init__(self, js_obj):
- super().__init__(js_obj.__dict__['_obj'])
+_THREAD_LOCAL_JS_CACHE = threading.local()
- def __getstate__(self):
- return self.__dict__.copy()
- def __setstate__(self, state):
- self.__dict__.update(state)
+class _JsWrapper:
+ def __init__(self, source_code, entrypoint_name):
+ self.source_code = source_code
+ self.entrypoint_name = entrypoint_name
+
+ def _get_wrapper_fn(self):
+ cache = _THREAD_LOCAL_JS_CACHE
+ if not hasattr(cache, 'functions'):
+ cache.functions = {}
+
+ cache_key = (self.source_code, self.entrypoint_name)
+ if cache_key not in cache.functions:
+ ctx = quickjs.Context()
+ ctx.eval(self.source_code)
+ cache.functions[cache_key] = ctx.get(self.entrypoint_name)
+
+ return cache.functions[cache_key]
+
+ def __call__(self, row):
+ wrapper_fn = self._get_wrapper_fn()
+ row_as_dict = py_value_to_js_dict(row)
+ try:
+ js_result = wrapper_fn(json.dumps(row_as_dict))
Review Comment:
updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
