@@ -69,31 +69,81 @@ def __init__(self, invocation_id):
6969 self .retry_context = _RetryCtx ()
7070
7171
72+ # --- Shared Datum-currency helpers -------------------------------------------
73+ #
74+ # The sync and async native paths differ ONLY in how the handler is executed.
75+ # Decoding inputs, preparing the invocation context, validating the return, and
76+ # encoding outputs are identical and protobuf-free, so they are factored out
77+ # here and reused by both entries below. (v1 has no deferred bindings, so
78+ # ``get_binding`` takes only the binding name.)
79+
80+
81+ def _decode_inputs (fi , inputs , trigger_metadata ):
82+ """Rust-provided ``(name, Datum)`` inputs -> decoded handler args dict."""
83+ metadata = trigger_metadata or {}
84+ args = {}
85+ for name , datum in inputs :
86+ pb_type_info = fi .input_types [name ]
87+ tm = metadata if is_trigger_binding (pb_type_info .binding_name ) else {}
88+ binding_obj = get_binding (pb_type_info .binding_name )
89+ args [name ] = binding_obj .decode (datum , trigger_metadata = tm )
90+ return args
91+
92+
93+ def _prepare_context (fi , invocation_id , args ):
94+ """Build the invocation context and scaffold ``Out`` params into *args*.
95+
96+ Does NOT stamp ``thread_local_storage.invocation_id`` -- the sync path lets
97+ ``run_sync_func`` do that on the executing thread, while the async path sets
98+ it explicitly on the caller's context (see below)."""
99+ fi_context = get_context (_InvocShim (invocation_id ), fi .name , fi .directory )
100+ if fi .requires_context :
101+ args ['context' ] = fi_context
102+ if fi .output_types :
103+ for name in fi .output_types :
104+ args [name ] = Out ()
105+ return fi_context
106+
107+
108+ def _check_return (fi , call_result ):
109+ if call_result is not None and not fi .has_return :
110+ raise RuntimeError (
111+ 'function %s without a $return binding returned a non-None value'
112+ % repr (fi .name ))
113+
114+
115+ def _collect_output (fi , args ):
116+ """Decoded ``Out`` params -> ``[(name, Datum)]`` for Rust to prost-encode."""
117+ output_data = []
118+ if fi .output_types :
119+ for out_name , out_type_info in fi .output_types .items ():
120+ val = args [out_name ].get ()
121+ if val is None :
122+ continue
123+ datum = get_datum (out_type_info .binding_name , val ,
124+ out_type_info .pytype )
125+ output_data .append ((out_name , datum ))
126+ return output_data
127+
128+
129+ def _encode_return (fi , call_result ):
130+ if fi .return_type is not None :
131+ return get_datum (fi .return_type .binding_name , call_result ,
132+ fi .return_type .pytype )
133+ return None
134+
135+
72136async def invocation_request_native (invocation_id , function_id , inputs ,
73137 trigger_metadata ):
74138 threadpool = get_threadpool_executor ()
75139 fi = _functions .get_function (function_id )
76140 if fi is None :
77141 return (False , None , [], "function %s not loaded" % function_id )
78142
79- metadata = trigger_metadata or {}
80- args = {}
81-
82143 try :
83- for name , datum in inputs :
84- pb_type_info = fi .input_types [name ]
85- tm = metadata if is_trigger_binding (pb_type_info .binding_name ) else {}
86- binding_obj = get_binding (pb_type_info .binding_name )
87- args [name ] = binding_obj .decode (datum , trigger_metadata = tm )
88-
89- fi_context = get_context (_InvocShim (invocation_id ), fi .name , fi .directory )
144+ args = _decode_inputs (fi , inputs , trigger_metadata )
145+ fi_context = _prepare_context (fi , invocation_id , args )
90146 fi_context .thread_local_storage .invocation_id = invocation_id
91- if fi .requires_context :
92- args ['context' ] = fi_context
93-
94- if fi .output_types :
95- for name in fi .output_types :
96- args [name ] = Out ()
97147
98148 if fi .is_async :
99149 if otel_manager .get_azure_monitor_available ():
@@ -115,25 +165,9 @@ async def invocation_request_native(invocation_id, function_id, inputs,
115165 threadpool , run_sync_func ,
116166 invocation_id , fi_context , fi .func , args )
117167
118- if call_result is not None and not fi .has_return :
119- raise RuntimeError (
120- 'function %s without a $return binding returned a non-None value'
121- % repr (fi .name ))
122-
123- output_data = []
124- if fi .output_types :
125- for out_name , out_type_info in fi .output_types .items ():
126- val = args [out_name ].get ()
127- if val is None :
128- continue
129- datum = get_datum (out_type_info .binding_name , val ,
130- out_type_info .pytype )
131- output_data .append ((out_name , datum ))
132-
133- return_datum = None
134- if fi .return_type is not None :
135- return_datum = get_datum (fi .return_type .binding_name , call_result ,
136- fi .return_type .pytype )
168+ _check_return (fi , call_result )
169+ output_data = _collect_output (fi , args )
170+ return_datum = _encode_return (fi , call_result )
137171
138172 sys .stdout .flush ()
139173 return (True , return_datum , output_data , None )
@@ -162,42 +196,14 @@ def run_invocation_sync(invocation_id, function_id, inputs, trigger_metadata):
162196 return (False , None , None , [], None )
163197
164198 try :
165- metadata = trigger_metadata or {}
166- args = {}
167- for name , datum in inputs :
168- pb_type_info = fi .input_types [name ]
169- tm = metadata if is_trigger_binding (pb_type_info .binding_name ) else {}
170- binding_obj = get_binding (pb_type_info .binding_name )
171- args [name ] = binding_obj .decode (datum , trigger_metadata = tm )
172-
173- fi_context = get_context (_InvocShim (invocation_id ), fi .name , fi .directory )
174- if fi .requires_context :
175- args ['context' ] = fi_context
176- if fi .output_types :
177- for name in fi .output_types :
178- args [name ] = Out ()
199+ args = _decode_inputs (fi , inputs , trigger_metadata )
200+ fi_context = _prepare_context (fi , invocation_id , args )
179201
180202 call_result = run_sync_func (invocation_id , fi_context , fi .func , args )
181203
182- if call_result is not None and not fi .has_return :
183- raise RuntimeError (
184- 'function %s without a $return binding returned a non-None value'
185- % repr (fi .name ))
186-
187- output_data = []
188- if fi .output_types :
189- for out_name , out_type_info in fi .output_types .items ():
190- val = args [out_name ].get ()
191- if val is None :
192- continue
193- datum = get_datum (out_type_info .binding_name , val ,
194- out_type_info .pytype )
195- output_data .append ((out_name , datum ))
196-
197- return_datum = None
198- if fi .return_type is not None :
199- return_datum = get_datum (fi .return_type .binding_name , call_result ,
200- fi .return_type .pytype )
204+ _check_return (fi , call_result )
205+ output_data = _collect_output (fi , args )
206+ return_datum = _encode_return (fi , call_result )
201207
202208 return (True , True , return_datum , output_data , None )
203209
0 commit comments