1717
1818"""This module defines the basic MapToFields operation."""
1919import itertools
20+ import json
2021import re
22+ import threading
2123from collections import abc
2224from collections .abc import Callable
2325from collections .abc import Collection
5355from apache_beam .yaml .yaml_errors import maybe_with_exception_handling_transform_fn
5456from apache_beam .yaml .yaml_provider import dicts_to_rows
5557
56- # Import js2py package if it exists
58+ # Import quickjs package if it exists
5759try :
58- import js2py
59- from js2py .base import JsObjectWrapper
60+ import quickjs
6061except ImportError :
61- js2py = None
62- JsObjectWrapper = object
62+ quickjs = None
6363
6464_str_expression_fields = {
6565 'AssignTimestamps' : 'timestamp' ,
@@ -178,18 +178,38 @@ def _check_mapping_arguments(
178178 raise ValueError (f'{ transform_name } cannot specify "name" without "path"' )
179179
180180
181- # js2py's JsObjectWrapper object has a self-referencing __dict__ property
182- # that cannot be pickled without implementing the __getstate__ and
183- # __setstate__ methods.
184- class _CustomJsObjectWrapper (JsObjectWrapper ):
185- def __init__ (self , js_obj ):
186- super ().__init__ (js_obj .__dict__ ['_obj' ])
181+ _THREAD_LOCAL_JS_CACHE = threading .local ()
187182
188- def __getstate__ (self ):
189- return self .__dict__ .copy ()
190183
191- def __setstate__ (self , state ):
192- self .__dict__ .update (state )
184+ class _JsWrapper :
185+ def __init__ (self , source_code , entrypoint_name ):
186+ self .source_code = source_code
187+ self .entrypoint_name = entrypoint_name
188+
189+ def _get_wrapper_fn (self ):
190+ cache = _THREAD_LOCAL_JS_CACHE
191+ if not hasattr (cache , 'functions' ):
192+ cache .functions = {}
193+
194+ cache_key = (self .source_code , self .entrypoint_name )
195+ if cache_key not in cache .functions :
196+ ctx = quickjs .Context ()
197+ ctx .eval (self .source_code )
198+ cache .functions [cache_key ] = ctx .get (self .entrypoint_name )
199+
200+ return cache .functions [cache_key ]
201+
202+ def __call__ (self , row ):
203+ wrapper_fn = self ._get_wrapper_fn ()
204+ row_as_dict = py_value_to_js_dict (row )
205+ try :
206+ js_result = wrapper_fn (json .dumps (row_as_dict ))
207+ except Exception as exn :
208+ raise RuntimeError (
209+ f"Error evaluating javascript expression: { exn } " ) from exn
210+ if hasattr (js_result , 'json' ):
211+ js_result = json .loads (js_result .json ())
212+ return dicts_to_rows (js_result )
193213
194214
195215# TODO(yaml) Improve type inferencing for JS UDF's
@@ -210,80 +230,54 @@ def py_value_to_js_dict(py_value):
210230def _expand_javascript_mapping_func (
211231 original_fields , expression = None , callable = None , path = None , name = None ):
212232
213- # Check for installed js2py package
214- if js2py is None :
233+ if quickjs is None :
215234 raise ValueError (
216- "Javascript mapping functions are not supported on"
217- " Python 3.12 or later." )
218-
219- # import remaining js2py objects
220- from js2py import base
221- from js2py .constructors import jsdate
222- from js2py .internals import simplex
223-
224- js_array_type = (
225- base .PyJsArray ,
226- base .PyJsArrayBuffer ,
227- base .PyJsInt8Array ,
228- base .PyJsUint8Array ,
229- base .PyJsUint8ClampedArray ,
230- base .PyJsInt16Array ,
231- base .PyJsUint16Array ,
232- base .PyJsInt32Array ,
233- base .PyJsUint32Array ,
234- base .PyJsFloat32Array ,
235- base .PyJsFloat64Array )
236-
237- def _js_object_to_py_object (obj ):
238- if isinstance (obj , (base .PyJsNumber , base .PyJsString , base .PyJsBoolean )):
239- return base .to_python (obj )
240- elif isinstance (obj , js_array_type ):
241- return [_js_object_to_py_object (value ) for value in obj .to_list ()]
242- elif isinstance (obj , jsdate .PyJsDate ):
243- return obj .to_utc_dt ()
244- elif isinstance (obj , (base .PyJsNull , base .PyJsUndefined )):
245- return None
246- elif isinstance (obj , base .PyJsError ):
247- raise RuntimeError (obj ['message' ])
248- elif isinstance (obj , base .PyJsObject ):
249- return {
250- key : _js_object_to_py_object (value ['value' ])
251- for (key , value ) in obj .own .items ()
252- }
253- elif isinstance (obj , base .JsObjectWrapper ):
254- return _js_object_to_py_object (obj ._obj )
255-
256- return obj
235+ "Javascript mapping functions are not supported because the "
236+ "quickjs-ng library is not installed." )
257237
258238 if expression :
259- source = '\n ' .join (['function(__row__) {' ] + [
260- f' { name } = __row__.{ name } '
261- for name in original_fields if name in expression
262- ] + [' return (' + expression + ')' ] + ['}' ])
263- js_func = _CustomJsObjectWrapper (js2py .eval_js (source ))
239+ unpacking_code = '\n ' .join ([
240+ f' var { name } = __row__.{ name } ;' for name in original_fields
241+ if name in expression
242+ ])
243+ source_code = f"""
244+ const udf = function(__row__) {{
245+ { unpacking_code }
246+ return ({ expression } );
247+ }};
248+ function wrapper(json_str) {{
249+ return udf(JSON.parse(json_str));
250+ }}
251+ """
252+ entrypoint = 'wrapper'
264253
265254 elif callable :
266- js_func = _CustomJsObjectWrapper (js2py .eval_js (callable ))
255+ match = re .search (r'(?:async\s+)?function\s+([a-zA-Z0-9_]+)' , callable )
256+ if not match :
257+ raise ValueError (
258+ f"Could not find function declaration in callable: { callable } " )
259+ udf_name = match .group (1 )
260+ source_code = f"""
261+ { callable }
262+ function wrapper(json_str) {{
263+ return { udf_name } (JSON.parse(json_str));
264+ }}
265+ """
266+ entrypoint = 'wrapper'
267267
268268 else :
269269 if not path .endswith ('.js' ):
270270 raise ValueError (f'File "{ path } " is not a valid .js file.' )
271271 udf_code = FileSystems .open (path ).read ().decode ()
272- js = js2py .EvalJs ()
273- js .eval (udf_code )
274- js_func = _CustomJsObjectWrapper (getattr (js , name ))
275-
276- def js_wrapper (row ):
277- row_as_dict = py_value_to_js_dict (row )
278- try :
279- js_result = js_func (row_as_dict )
280- except simplex .JsException as exn :
281- raise RuntimeError (
282- f"Error evaluating javascript expression: "
283- f"{ exn .mes ['message' ]} " ) from exn
284- return dicts_to_rows (_js_object_to_py_object (js_result ))
285-
286- return js_wrapper
272+ source_code = f"""
273+ { udf_code }
274+ function wrapper(json_str) {{
275+ return { name } (JSON.parse(json_str));
276+ }}
277+ """
278+ entrypoint = 'wrapper'
279+
280+ return _JsWrapper (source_code , entrypoint )
287281
288282
289283def _expand_python_mapping_func (
0 commit comments