1616#
1717
1818"""This module defines the basic MapToFields operation."""
19+ import datetime
1920import itertools
21+ import json
2022import re
23+ import threading
2124from collections import abc
2225from collections .abc import Callable
2326from collections .abc import Collection
2427from collections .abc import Iterable
2528from collections .abc import Mapping
29+ from decimal import Decimal
2630from typing import Any
2731from typing import NamedTuple
2832from typing import Optional
5357from apache_beam .yaml .yaml_errors import maybe_with_exception_handling_transform_fn
5458from apache_beam .yaml .yaml_provider import dicts_to_rows
5559
56- # Import js2py package if it exists
60+ # Import quickjs package if it exists
5761try :
58- import js2py
59- from js2py .base import JsObjectWrapper
62+ import quickjs
6063except ImportError :
61- js2py = None
62- JsObjectWrapper = object
64+ quickjs = None
6365
6466_str_expression_fields = {
6567 'AssignTimestamps' : 'timestamp' ,
@@ -178,18 +180,52 @@ def _check_mapping_arguments(
178180 raise ValueError (f'{ transform_name } cannot specify "name" without "path"' )
179181
180182
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' ])
183+ _THREAD_LOCAL_JS_CACHE = threading .local ()
187184
188- def __getstate__ (self ):
189- return self .__dict__ .copy ()
190185
191- def __setstate__ (self , state ):
192- self .__dict__ .update (state )
186+ class _JsFunctionWrapper :
187+ def __init__ (self , source_code , entrypoint_name ):
188+ self .source_code = source_code
189+ self .entrypoint_name = entrypoint_name
190+
191+ def _get_fn (self ):
192+ cache = _THREAD_LOCAL_JS_CACHE
193+ if not hasattr (cache , 'functions' ):
194+ cache .functions = {}
195+
196+ cache_key = (self .source_code , self .entrypoint_name )
197+ if cache_key not in cache .functions :
198+ context = quickjs .Context ()
199+ context .eval (self .source_code )
200+ f = context .get (self .entrypoint_name )
201+
202+ def call_fn (* args , run_gc = True ):
203+ def convert_arg (arg ):
204+ if isinstance (arg , (type (None ), str , bool , float , int )):
205+ return arg
206+ else :
207+ return context .parse_json (json .dumps (arg ))
208+
209+ try :
210+ result = f (* [convert_arg (a ) for a in args ])
211+ if isinstance (result , quickjs .Object ):
212+ result = json .loads (result .json ())
213+ return result
214+ finally :
215+ if run_gc :
216+ context .gc ()
217+
218+ cache .functions [cache_key ] = call_fn
219+
220+ return cache .functions [cache_key ]
221+
222+ def __call__ (self , row ):
223+ fn = self ._get_fn ()
224+ try :
225+ return dicts_to_rows (fn (py_value_to_js_dict (row )))
226+ except Exception as exn :
227+ raise RuntimeError (
228+ f"Error evaluating javascript expression: { exn } " ) from exn
193229
194230
195231# TODO(yaml) Improve type inferencing for JS UDF's
@@ -199,6 +235,12 @@ def py_value_to_js_dict(py_value):
199235 py_value = py_value ._asdict ()
200236 if isinstance (py_value , dict ):
201237 return {key : py_value_to_js_dict (value ) for key , value in py_value .items ()}
238+ elif isinstance (py_value , bytes ):
239+ return py_value .decode ('utf-8' , errors = 'replace' )
240+ elif isinstance (py_value , (datetime .datetime , datetime .date , datetime .time )):
241+ return {'__date__' : True , 'value' : py_value .isoformat ()}
242+ elif isinstance (py_value , Decimal ):
243+ return float (py_value )
202244 elif not isinstance (py_value , str ) and isinstance (py_value , abc .Iterable ):
203245 return [py_value_to_js_dict (value ) for value in list (py_value )]
204246 else :
@@ -210,80 +252,53 @@ def py_value_to_js_dict(py_value):
210252def _expand_javascript_mapping_func (
211253 original_fields , expression = None , callable = None , path = None , name = None ):
212254
213- # Check for installed js2py package
214- if js2py is None :
255+ if quickjs is None :
215256 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
257+ "Javascript mapping functions are not supported because the "
258+ "quickjs-ng library is not installed." )
257259
258260 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 ))
261+ source_code = f"""
262+ function udf(__row__) {{
263+ with (__row__) {{
264+ return ({ expression } );
265+ }}
266+ }}
267+ """
268+ user_entrypoint = 'udf'
264269
265270 elif callable :
266- js_func = _CustomJsObjectWrapper (js2py .eval_js (callable ))
271+ source_code = f"var __udf__ = ({ callable } );"
272+ user_entrypoint = '__udf__'
267273
268274 else :
269275 if not path .endswith ('.js' ):
270276 raise ValueError (f'File "{ path } " is not a valid .js file.' )
271277 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 ))
278+ source_code = udf_code
279+ user_entrypoint = name
280+
281+ source_code += f"""
282+ function __convert_dates__(obj) {{
283+ if (obj && typeof obj === 'object') {{
284+ if (obj.__date__) {{
285+ return new Date(obj.value);
286+ }}
287+ for (var key in obj) {{
288+ if (obj.hasOwnProperty(key)) {{
289+ obj[key] = __convert_dates__(obj[key]);
290+ }}
291+ }}
292+ }}
293+ return obj;
294+ }}
295+
296+ function __wrapper__(row) {{
297+ return { user_entrypoint } (__convert_dates__(row));
298+ }}
299+ """
285300
286- return js_wrapper
301+ return _JsFunctionWrapper ( source_code , '__wrapper__' )
287302
288303
289304def _expand_python_mapping_func (
0 commit comments