refactor: hold planner contexts natively and call native helpers directly

crates/sqlbuild-python/src/bindings/_helpers/analysis/explicit_references.rs

@@ -386,7 +386,7 @@ fn python_literal_sql_calls(
 /// How an owner of `owner_kind` should read `target` instead of naming its relation in SQL.
 #[pyfunction]
 #[pyo3(signature = (*, owner_kind, r#ref, upstream_loader_by_source = None))]
-fn hard_coded_relation_remedy(
+pub(crate) fn hard_coded_relation_remedy(
     owner_kind: &str,
     r#ref: &Bound<'_, PyAny>,
     upstream_loader_by_source: Option<&Bound<'_, PyAny>>,
@@ -407,7 +407,7 @@ fn hard_coded_relation_remedy(
 /// The P008 message for SQL naming a relation kept only as a compatibility view.
 #[pyfunction]
 #[pyo3(signature = (*, owner_label, written, method, model_name))]
-fn old_name_reference_message(
+pub(crate) fn old_name_reference_message(
     owner_label: &str,
     written: &str,
     method: &str,

crates/sqlbuild-python/src/bindings/_helpers/auditing/mod.rs

@@ -1,5 +1,5 @@
 mod identity;
 mod measurement;
 pub(crate) mod registration;
-mod render;
+pub(crate) mod render;
 mod schema_audits;

crates/sqlbuild-python/src/bindings/_helpers/auditing/render.rs

@@ -30,7 +30,7 @@ fn qualified<'py>(mapping: &Bound<'py, PyAny>, name: &str) -> PyResult<Option<Bo
 /// Render audit SQL from unresolved markers, applying relation overrides first.
 #[pyfunction]
 #[pyo3(signature = (**request))]
-fn render_audit_sql(request: Option<&Bound<'_, PyDict>>) -> PyResult<String> {
+pub(crate) fn render_audit_sql(request: Option<&Bound<'_, PyDict>>) -> PyResult<String> {
     let request = Request::new("render_audit_sql", request)?;
     let sql: String = request.get("unresolved_sql")?.extract()?;
     let models: Bound<'_, PyAny> = request.get("model_locations")?;

crates/sqlbuild-python/src/bindings/_helpers/authored_values/project_vars.rs

@@ -45,7 +45,7 @@ fn preview_json(value: &Bound<'_, PyAny>) -> PyResult<String> {
 /// One project var value as interpolated text; structured values are rejected.
 #[pyfunction]
 #[pyo3(signature = (*, value, label))]
-fn render_project_var_text(value: &Bound<'_, PyAny>, label: &str) -> PyResult<String> {
+pub(crate) fn render_project_var_text(value: &Bound<'_, PyAny>, label: &str) -> PyResult<String> {
     project_var_text(value, label)
 }
 

crates/sqlbuild-python/src/bindings/_helpers/check/check_command.rs

@@ -1,5 +1,21 @@
 //! `sqb check`: run the selected Python checks after their Python dependencies.
 
+use crate::bindings::_helpers::cli::progress::current_native_progress_projector;
+use crate::bindings::_helpers::cli::progress_text::write_execution_header;
+use crate::bindings::_helpers::compile_driver::pipeline_entries::build_static_pipeline_result;
+use crate::bindings::_helpers::compile_driver::python_calls::keywords;
+use crate::bindings::_helpers::compile_driver::relation_targets::build_project_relation_targets;
+use crate::bindings::_helpers::compile_driver::relation_targets::build_python_relation_targets;
+use crate::bindings::_helpers::compile_driver::selection_phases::compile_static_command_context;
+use crate::bindings::_helpers::dbt_interop::reference_resolver::resolve_external_sql_reference_resolver;
+use crate::bindings::_helpers::graph::algorithms::transitive_closure_many;
+use crate::bindings::_helpers::graph::execution_gates::sql_ref_key;
+use crate::bindings::_helpers::graph::python_lifecycle::build_python_sql_run_lifecycle;
+use crate::bindings::_helpers::planner_entries::command_plans::build_relation_command_plan;
+use crate::bindings::_helpers::planner_phases::objects::object;
+use crate::bindings::_helpers::planner_selection::selectors::parse_project_selector;
+use crate::bindings::_helpers::planner_selection::unit_tests::reject_unit_test_selectors;
+use crate::bindings::_helpers::result_output::documents::write_execution_json_output;
 use pyo3::prelude::{
     Bound, Py, PyAny, PyAnyMethods, PyErr, PyModule, PyModuleMethods, PyResult, Python,
 };
@@ -70,12 +86,12 @@ fn empty_tuple(py: Python<'_>) -> Bound<'_, PyAny> {
 
 /// Refuse `+`, `~` and path selectors, splitting selectors as Python's `split` does.
 fn refuse_graph_selectors(py: Python<'_>, selectors: &[Bound<'_, PyAny>]) -> PyResult<()> {
-    let parse: Bound<'_, PyAny> = registry::native(py, NativeAttr::ParseProjectSelector)?;
     let path_selector: Bound<'_, PyAny> = registry::class(py, RecordId::PathSelector)?;
     for selector in selectors {
         for token in selector.call_method0("split")?.try_iter()? {
             for part in token?.call_method1("split", (",",))?.try_iter()? {
-                let parsed: Bound<'_, PyAny> = parse.call1((part?,))?;
+                let parsed: Bound<'_, PyAny> =
+                    parse_project_selector(part?.downcast::<PyString>()?)?;
                 if parsed.is_instance(&path_selector)?
                     || parsed.getattr("upstream")?.is_truthy()?
                     || parsed.getattr("downstream")?.is_truthy()?
@@ -163,16 +179,14 @@ fn compile_pipeline<'py>(
         &graph.getattr("selected_sql_refs")?,
         &[("selected_names", check_names.clone().into_any())],
     )?;
-    let closure: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::TransitiveClosureMany)?,
-        &[
-            ("starts", check_names.clone().into_any()),
-            ("edges", graph.getattr("upstream_deps")?),
-            (
-                "include_starts",
-                PyBool::new(py, false).to_owned().into_any(),
-            ),
-        ],
+    let closure: Bound<'py, PyAny> = object(
+        py,
+        transitive_closure_many(
+            check_names.as_any(),
+            &graph.getattr("upstream_deps")?,
+            false,
+            None,
+        )?,
     )?;
     let loader_kind: Bound<'py, PyAny> =
         host::entry(py, UserCodeHost::PythonNodeKind)?.getattr("LOADER")?;
@@ -187,10 +201,9 @@ fn compile_pipeline<'py>(
     let key_class: Bound<'py, PyAny> = registry::class(py, RecordId::CompiledObjectKey)?;
     let source_type: Bound<'py, PyAny> =
         imported(py, COMPILE_TYPES, "CompiledResourceType")?.getattr("SOURCE")?;
-    let ref_key: Bound<'py, PyAny> = registry::native(py, NativeAttr::SqlRefKey)?;
     let mut keys: Vec<Bound<'py, PyAny>> = Vec::new();
     for reference in required_refs.try_iter()? {
-        keys.push(ref_key.call1((reference?,))?);
+        keys.push(sql_ref_key(py, &reference?)?);
     }
     for source_file in discovered.getattr("source_files")?.try_iter()? {
         for source in source_file?.getattr("source_entries")?.try_iter()? {
@@ -207,12 +220,9 @@ fn compile_pipeline<'py>(
         }
     }
     let keys: Bound<'py, PyAny> = PyFrozenSet::new(py, keys)?.into_any();
-    let resolver: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::ResolveExternalSqlReferenceResolver)?,
-        &[
-            ("project_dir", invocation.getattr("effective_project_dir")?),
-            ("discovered_inputs", discovered.clone()),
-        ],
+    let resolver: Bound<'py, PyAny> = resolve_external_sql_reference_resolver(
+        invocation.getattr("effective_project_dir")?,
+        discovered.clone(),
     )?;
     let options: Bound<'py, PyAny> = call_with(
         &registry::class(py, RecordId::CompilePipelineOptions)?,
@@ -227,8 +237,8 @@ fn compile_pipeline<'py>(
             ("external_sql_reference_resolver", resolver),
         ],
     )?;
-    let context: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::CompileStaticCommandContext)?,
+    let context: Bound<'py, PyAny> = compile_static_command_context(Some(&keywords(
+        py,
         &[
             ("discovered_inputs", discovered),
             ("adapter", invocation.getattr("adapter")?),
@@ -242,18 +252,18 @@ fn compile_pipeline<'py>(
                     .getattr("on_progress")?,
             ),
         ],
+    )?))?;
+    let plan_output: Bound<'py, PyAny> = build_relation_command_plan(
+        &context.getattr("project")?,
+        &context.getattr("scope")?,
+        &context.getattr("relations")?,
     )?;
-    let plan_output: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::BuildRelationCommandPlan)?,
-        &[
-            ("project", context.getattr("project")?),
-            ("scope", context.getattr("scope")?),
-            ("relations", context.getattr("relations")?),
-        ],
-    )?;
-    call_with(
-        &registry::native(py, NativeAttr::BuildStaticPipelineResult)?,
-        &[("context", context), ("plan_output", plan_output)],
+    build_static_pipeline_result(
+        py,
+        Some(keywords(
+            py,
+            &[("context", context), ("plan_output", plan_output)],
+        )?),
     )
 }
 
@@ -275,15 +285,12 @@ fn python_selection<'py>(
     request: &Bound<'py, PyAny>,
     discovered: &Bound<'py, PyAny>,
 ) -> PyResult<(Bound<'py, PyAny>, Bound<'py, PyFrozenSet>)> {
-    let py: Python<'py> = request.py();
+    let _py: Python<'py> = request.py();
     let python_graph: Bound<'py, PyAny> = python_node_graph(discovered)?;
-    call_with(
-        &registry::native(py, NativeAttr::RejectUnitTestSelectors)?,
-        &[
-            ("select", request.getattr("select")?),
-            ("exclude", request.getattr("exclude")?),
-            ("discovered_inputs", discovered.clone()),
-        ],
+    reject_unit_test_selectors(
+        &request.getattr("select")?,
+        &request.getattr("exclude")?,
+        discovered,
     )?;
     let check_names: Bound<'py, PyFrozenSet> = selected_check_names(request, &python_graph)?;
     Ok((python_graph, check_names))
@@ -314,27 +321,28 @@ fn prepare<'py>(
             ("python_node_names", PyFrozenSet::empty(py)?.into_any()),
         ],
     )?;
-    let lifecycle_plan: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::BuildPythonSqlRunLifecycle)?,
-        &[
-            ("selection", selection),
-            ("python_graph", python_graph.clone()),
-        ],
-    )?;
+    let lifecycle_plan: Bound<'py, PyAny> =
+        build_python_sql_run_lifecycle(&selection, &python_graph)?;
     let relation_refs: Bound<'py, PyAny> = call_with(
         &python_graph.getattr("selected_sql_refs")?,
         &[("selected_names", check_names.clone().into_any())],
     )?;
     let project: Bound<'py, PyAny> = pipeline.getattr("project")?;
     let plan_output: Bound<'py, PyAny> = pipeline.getattr("plan_output")?;
-    let relation_targets: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::BuildPythonRelationTargets)?,
-        &[
-            ("adapter", adapter.clone()),
-            ("project", project.clone()),
-            ("plan_output", plan_output.clone()),
-            ("required_refs", relation_refs.clone()),
-        ],
+    let relation_targets: Bound<'py, PyAny> = object(
+        py,
+        build_python_relation_targets(
+            py,
+            Some(keywords(
+                py,
+                &[
+                    ("adapter", adapter.clone()),
+                    ("project", project.clone()),
+                    ("plan_output", plan_output.clone()),
+                    ("required_refs", relation_refs.clone()),
+                ],
+            )?),
+        )?,
     )?;
     Ok(Prepared {
         check_functions: PyTuple::new(py, check_functions)?.into_any(),
@@ -353,9 +361,15 @@ fn prepare<'py>(
                 ("effective_target_schema", "default_schema"),
             )?,
         ),
-        project_relations: call_with(
-            &registry::native(py, NativeAttr::BuildProjectRelationTargets)?,
-            &[("adapter", adapter), ("plan_output", plan_output)],
+        project_relations: object(
+            py,
+            build_project_relation_targets(
+                py,
+                Some(keywords(
+                    py,
+                    &[("adapter", adapter), ("plan_output", plan_output)],
+                )?),
+            )?,
         )?,
         python_graph,
         pipeline,
@@ -556,8 +570,7 @@ fn execute<'py>(
     let event_writer: Bound<'py, PyAny> =
         registry::native(py, NativeAttr::ExecutionEventWriter)?.call0()?;
     let executed: PyResult<Vec<Bound<'py, PyAny>>> = (|| {
-        let projector: Bound<'py, PyAny> =
-            registry::native(py, NativeAttr::CurrentNativeProgressProjector)?.call0()?;
+        let projector: Bound<'py, PyAny> = current_native_progress_projector(py)?;
         if !projector.is_none() {
             let ordinals: Bound<'py, PyDict> = PyDict::new(py);
             let mut total: usize = 0;
@@ -617,16 +630,14 @@ fn complete<'py>(
             .call_method1("__truediv__", ("target",))?,
         &tuple,
     )?;
-    call_with(
-        &registry::native(py, NativeAttr::WriteExecutionJsonOutput)?,
-        &[
-            (
-                "payload",
-                PyString::new(py, &results_json(results)?).into_any(),
-            ),
-            ("json_output", request.getattr("json_output")?),
-            ("json_output_path", request.getattr("json_output_path")?),
-        ],
+    object(
+        py,
+        write_execution_json_output(
+            py,
+            PyString::new(py, &results_json(results)?).into_any(),
+            request.getattr("json_output")?.is_truthy()?,
+            request.getattr("json_output_path")?,
+        )?,
     )?;
     Ok(check_exit_code(&texts))
 }
@@ -648,7 +659,7 @@ fn run_check_command(py: Python<'_>, request: &Bound<'_, PyAny>) -> PyResult<i32
     header.set_item("target", py.None())?;
     header.set_item("concurrency", PyInt::new(py, 1))?;
     header.set_item("use_color", invocation.getattr("use_color")?)?;
-    registry::native(py, NativeAttr::WriteExecutionHeader)?.call((), Some(&header))?;
+    write_execution_header(py, Some(header))?;
     let prepared: Prepared<'_> = prepare(request, &invocation)?;
     let session: Bound<'_, PyAny> = call_with(
         &host::entry(py, UserCodeHost::BuildProviderSession)?,

crates/sqlbuild-python/src/bindings/_helpers/check/check_core.rs

@@ -1,5 +1,9 @@
 //! Python check planning: run-state recording, dependency relations and read-side preflight.
 
+use crate::bindings::_helpers::compile_driver::python_calls::keywords;
+use crate::bindings::_helpers::compile_driver::relation_targets::build_python_relation_targets;
+use crate::bindings::_helpers::graph::python_selection::resolve_python_check_excludes;
+use crate::bindings::_helpers::planner_phases::objects::object;
 use pyo3::IntoPyObject;
 use pyo3::prelude::{
     Bound, PyAny, PyAnyMethods, PyErr, PyModule, PyModuleMethods, PyResult, Python,
@@ -15,8 +19,6 @@ use crate::bindings::_helpers::native_host::adapter::AdapterHost;
 use crate::bindings::_helpers::native_host::executor::ExecutorHost;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::native_host::user_code::UserCodeHost;
-use crate::bindings::_helpers::records::ids::NativeAttr;
-use crate::bindings::_helpers::records::registry;
 
 const MISSING_RELATION_CODE: &str = "C682";
 const MODEL_KIND: &str = "model";
@@ -170,14 +172,20 @@ pub(crate) fn build_check_relation_targets<'py>(
         &python_graph.getattr("selected_sql_refs")?,
         &[("selected_names", selected_python_names)],
     )?;
-    call_with(
-        &registry::native(py, NativeAttr::BuildPythonRelationTargets)?,
-        &[
-            ("adapter", adapter.clone()),
-            ("project", pipeline_result.getattr("project")?),
-            ("plan_output", pipeline_result.getattr("plan_output")?),
-            ("required_refs", required),
-        ],
+    object(
+        py,
+        build_python_relation_targets(
+            py,
+            Some(keywords(
+                py,
+                &[
+                    ("adapter", adapter.clone()),
+                    ("project", pipeline_result.getattr("project")?),
+                    ("plan_output", pipeline_result.getattr("plan_output")?),
+                    ("required_refs", required),
+                ],
+            )?),
+        )?,
     )
 }
 
@@ -189,9 +197,9 @@ pub(crate) fn relevant_check_functions<'py>(
     (selected_dependency_names, completed_sql_names): (&Bound<'py, PyAny>, &Bound<'py, PyAny>),
 ) -> PyResult<Vec<Bound<'py, PyAny>>> {
     let py: Python<'py> = discovered_inputs.py();
-    let excluded: Bound<'py, PyAny> = call_with(
-        &registry::native(py, NativeAttr::ResolvePythonCheckExcludes)?,
-        &[("graph", python_graph.clone()), ("exclude", exclude)],
+    let excluded: Bound<'py, PyAny> = object(
+        py,
+        resolve_python_check_excludes(python_graph, exclude.extract::<Vec<String>>()?)?,
     )?;
     let check_kind: Bound<'py, PyAny> =
         host::entry(py, UserCodeHost::PythonNodeKind)?.getattr("CHECK")?;

crates/sqlbuild-python/src/bindings/_helpers/check/check_results.rs

@@ -14,10 +14,10 @@ use sqlbuild_core::check_command::models::{CheckGrouping, CheckResultText, Check
 use sqlbuild_core::cli_text::models::CliStyle;
 use sqlbuild_core::json::models::JsonValue;
 
+use crate::bindings::_helpers::cli::progress::current_native_progress_projector;
 use crate::bindings::_helpers::compile_driver::python_calls::KeywordArguments;
-use crate::bindings::_helpers::records::ids::NativeAttr;
-use crate::bindings::_helpers::records::registry;
 use crate::bindings::_helpers::result_output::integration_values::json_value;
+use crate::bindings::_helpers::result_output::terminal_index::current_terminal_event_index;
 
 const WARN_SEVERITY: &str = "warn";
 
@@ -124,8 +124,7 @@ pub(crate) fn write_results(
         let check: Bound<'_, PyAny> = check?;
         checks.insert(check.getattr("name")?.extract()?, check);
     }
-    let projector: Bound<'_, PyAny> =
-        registry::native(py, NativeAttr::CurrentNativeProgressProjector)?.call0()?;
+    let projector: Bound<'_, PyAny> = current_native_progress_projector(py)?;
     let mut groups: Vec<(String, Vec<CheckResultText>)> = Vec::new();
     for result in results {
         let text: CheckResultText = check_result_text(result)?;
@@ -169,8 +168,7 @@ pub(crate) fn results_json(results: &[Bound<'_, PyAny>]) -> PyResult<String> {
         return check_json(&[]).map_err(|error| PyValueError::new_err(format!("{error:?}")));
     };
     let py: Python<'_> = first.py();
-    let index: Bound<'_, PyAny> =
-        registry::native(py, NativeAttr::CurrentTerminalEventIndex)?.call0()?;
+    let index: Bound<'_, PyAny> = current_terminal_event_index(py)?;
     let mut texts: Vec<CheckResultText> = Vec::new();
     for result in results {
         let name: Bound<'_, PyAny> = result.getattr("node_name")?;