chore: always pass PyExecutionContext to the analyzer (#1195)

Author: georgeh0

python/cocoindex/flow.py

@@ -459,7 +459,9 @@ class _FlowBuilderState:
     field_name_builder: _NameBuilder
 
     def __init__(self, full_name: str):
-        self.engine_flow_builder = _engine.FlowBuilder(full_name)
+        self.engine_flow_builder = _engine.FlowBuilder(
+            full_name, execution_context.event_loop
+        )
         self.field_name_builder = _NameBuilder()
 
     def get_data_slice(self, v: Any) -> _engine.DataSlice:
@@ -931,9 +933,7 @@ def _create_engine_flow() -> _engine.Flow:
             flow_builder_state, flow_builder_state.engine_flow_builder.root_scope()
         )
         fl_def(FlowBuilder(flow_builder_state), root_scope)
-        return flow_builder_state.engine_flow_builder.build_flow(
-            execution_context.event_loop
-        )
+        return flow_builder_state.engine_flow_builder.build_flow()
 
     return Flow(flow_name, _create_engine_flow)
 

src/builder/analyzed_flow.rs

@@ -58,7 +58,8 @@ impl AnalyzedTransientFlow {
         transient_flow: spec::TransientFlowSpec,
         py_exec_ctx: Option<crate::py::PythonExecutionContext>,
     ) -> Result<Self> {
-        let ctx = analyzer::build_flow_instance_context(&transient_flow.name, py_exec_ctx);
+        let ctx =
+            analyzer::build_flow_instance_context(&transient_flow.name, py_exec_ctx.map(Arc::new));
         let (output_type, data_schema, execution_plan_fut) =
             analyzer::analyze_transient_flow(&transient_flow, ctx).await?;
         Ok(Self {

src/builder/analyzer.rs

@@ -1033,12 +1033,12 @@ impl AnalyzerContext {
 
 pub fn build_flow_instance_context(
     flow_inst_name: &str,
-    py_exec_ctx: Option<crate::py::PythonExecutionContext>,
+    py_exec_ctx: Option<Arc<crate::py::PythonExecutionContext>>,
 ) -> Arc<FlowInstanceContext> {
     Arc::new(FlowInstanceContext {
         flow_instance_name: flow_inst_name.to_string(),
         auth_registry: get_auth_registry().clone(),
-        py_exec_ctx: py_exec_ctx.map(Arc::new),
+        py_exec_ctx: py_exec_ctx,
     })
 }
 

src/builder/flow_builder.rs

@@ -247,7 +247,7 @@ pub struct FlowBuilder {
 #[pymethods]
 impl FlowBuilder {
     #[new]
-    pub fn new(py: Python<'_>, name: &str) -> PyResult<Self> {
+    pub fn new(py: Python<'_>, name: &str, py_event_loop: Py<PyAny>) -> PyResult<Self> {
         let lib_context = py
             .allow_threads(|| -> anyhow::Result<Arc<LibContext>> {
                 get_runtime().block_on(get_lib_context())
@@ -258,7 +258,13 @@ impl FlowBuilder {
             None,
             Arc::new(Mutex::new(DataScopeBuilder::new())),
         );
-        let flow_inst_context = build_flow_instance_context(name, None);
+        let flow_inst_context = build_flow_instance_context(
+            name,
+            Some(Arc::new(crate::py::PythonExecutionContext::new(
+                py,
+                py_event_loop,
+            ))),
+        );
         let result = Self {
             lib_context,
             flow_inst_context,
@@ -606,18 +612,15 @@ impl FlowBuilder {
         }))
     }
 
-    pub fn build_flow(&self, py: Python<'_>, py_event_loop: Py<PyAny>) -> PyResult<py::Flow> {
+    pub fn build_flow(&self, py: Python<'_>) -> PyResult<py::Flow> {
         let spec = spec::FlowInstanceSpec {
             name: self.flow_instance_name.clone(),
             import_ops: self.import_ops.clone(),
             reactive_ops: self.reactive_ops.clone(),
             export_ops: self.export_ops.clone(),
             declarations: self.declarations.clone(),
         };
-        let flow_instance_ctx = build_flow_instance_context(
-            &self.flow_instance_name,
-            Some(crate::py::PythonExecutionContext::new(py, py_event_loop)),
-        );
+        let flow_instance_ctx = self.flow_inst_context.clone();
         let flow_ctx = py
             .allow_threads(|| {
                 get_runtime().block_on(async move {