refactor(native): read planner config through typed values

crates/sqlbuild-python/src/bindings/_helpers/model_config/typed_config.rs

@@ -5,9 +5,14 @@ use std::sync::{Arc, Mutex};
 use pyo3::exceptions::PyRuntimeError;
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyErr, PyResult, Python};
 use pyo3::pyclass;
+use pyo3::types::{PyString, PyTuple};
 use sqlbuild_model_config::config_values::errors::ConfigTypeError;
+use sqlbuild_model_config::config_values::main::checked_cursor_bound::checked_cursor_bound;
+use sqlbuild_model_config::config_values::main::checked_flag::checked_flag;
+use sqlbuild_model_config::config_values::main::checked_integer::checked_integer;
 use sqlbuild_model_config::config_values::main::checked_text::checked_text;
-use sqlbuild_model_config::config_values::models::ConfigNode;
+use sqlbuild_model_config::config_values::main::checked_texts::checked_texts;
+use sqlbuild_model_config::config_values::models::{ConfigNode, CursorBound};
 
 use crate::bindings::_helpers::compile_driver::stage_work::{TYPED_CONFIG_READS, count_stage};
 use crate::bindings::_helpers::model_config::config_nodes::{
@@ -21,6 +26,7 @@ use crate::bindings::_helpers::records::models::NativeRecord;
 
 const CONFIG_FIELD: &str = "config";
 const VALUES_FIELD: &str = "values";
+const ISOFORMAT_METHOD: &str = "isoformat";
 /// The config record's cache slot holding its typed values; no Python attribute reads it.
 const TYPED_VALUES_SLOT: &str = "__typed_values__";
 const POISONED: &str = "typed config values cache is poisoned";
@@ -161,3 +167,87 @@ pub(crate) fn config_text_is(
         _ => false,
     })
 }
+
+/// A cursor bound's text object: a string's or integer's `str()`, a date's `isoformat()`.
+pub(crate) fn cursor_bound_object<'py>(
+    py: Python<'py>,
+    bound: CursorBound<'_, HostHandle>,
+) -> PyResult<Bound<'py, PyAny>> {
+    match bound {
+        CursorBound::Dated(value) => config_object(py, value)?.call_method0(ISOFORMAT_METHOD),
+        CursorBound::Stringified(ConfigNode::Str(text)) => Ok(PyString::new(py, text).into_any()),
+        CursorBound::Stringified(ConfigNode::Int(number)) => {
+            Ok(PyString::new(py, &number.to_string()).into_any())
+        }
+        CursorBound::Stringified(value) => Ok(config_object(py, value)?.str()?.into_any()),
+    }
+}
+
+/// `key`'s cursor bound as text, or `None` when absent or `None`.
+pub(crate) fn config_cursor_bound<'py>(
+    py: Python<'py>,
+    values: &ConfigValues<'_>,
+    key: &str,
+) -> PyResult<Option<Bound<'py, PyAny>>> {
+    let node: Arc<PyConfigNode> = node(values, key)?;
+    checked_cursor_bound(key, Some(&node))
+        .map_err(|error| type_error(py, &error))?
+        .map(|bound| cursor_bound_object(py, bound))
+        .transpose()
+}
+
+/// `values.get(key)` as its Python object, `None` when absent.
+pub(crate) fn config_value<'py>(
+    py: Python<'py>,
+    values: &ConfigValues<'_>,
+    key: &str,
+) -> PyResult<Bound<'py, PyAny>> {
+    let value: Arc<PyConfigNode> = node(values, key)?;
+    config_object(py, &value)
+}
+
+/// `values.get(key)` as its typed node, `None` (null) when absent.
+pub(crate) fn config_node_of(values: &ConfigValues<'_>, key: &str) -> PyResult<Arc<PyConfigNode>> {
+    node(values, key)
+}
+
+/// `key`'s boolean value, or `None` when absent or `None`.
+pub(crate) fn config_flag(
+    py: Python<'_>,
+    values: &ConfigValues<'_>,
+    key: &str,
+) -> PyResult<Option<bool>> {
+    let node: Arc<PyConfigNode> = node(values, key)?;
+    checked_flag(key, Some(&node)).map_err(|error| type_error(py, &error))
+}
+
+/// `key`'s integer value as its Python object, or `None` when absent or `None`.
+pub(crate) fn config_integer_object<'py>(
+    py: Python<'py>,
+    values: &ConfigValues<'_>,
+    key: &str,
+) -> PyResult<Option<Bound<'py, PyAny>>> {
+    let node: Arc<PyConfigNode> = node(values, key)?;
+    checked_integer(key, Some(&node))
+        .map_err(|error| type_error(py, &error))?
+        .map(|value| config_object(py, value))
+        .transpose()
+}
+
+/// `key`'s strings as a tuple, or `None` when absent or `None`.
+pub(crate) fn config_text_tuple<'py>(
+    py: Python<'py>,
+    values: &ConfigValues<'_>,
+    key: &str,
+) -> PyResult<Option<Bound<'py, PyTuple>>> {
+    let node: Arc<PyConfigNode> = node(values, key)?;
+    let Some(items) = checked_texts(key, Some(&node)).map_err(|error| type_error(py, &error))?
+    else {
+        return Ok(None);
+    };
+    let objects: Vec<Bound<'py, PyAny>> = items
+        .into_iter()
+        .map(|item| config_object(py, item))
+        .collect::<PyResult<_>>()?;
+    PyTuple::new(py, objects).map(Some)
+}

crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs

@@ -4,6 +4,7 @@ use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::graph::project_graph::NativeProjectGraph;
+use crate::bindings::_helpers::model_config::typed_config::{config_node_of, model_values};
 use crate::bindings::_helpers::planner_entries::models::ModelPlanContext;
 use crate::bindings::_helpers::planner_entries::values::{
     native, planner_model, planner_type, string,
@@ -28,17 +29,21 @@ use crate::bindings::_helpers::records::fields::field as native_field;
 use crate::bindings::_helpers::records::ids::RecordId;
 use pyo3::PyRef;
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyModule, PyModuleMethods, PyResult, Python};
-use pyo3::types::{PyBool, PyDict, PyDictMethods, PyFrozenSet, PySet, PySetMethods, PyTuple};
+use pyo3::types::{PyDict, PyDictMethods, PyFrozenSet, PySet, PySetMethods, PyTuple};
 use pyo3::{pyfunction, wrap_pyfunction};
+use sqlbuild_model_config::config_values::models::ConfigNode;
+
+const ENABLED_KEY: &str = "enabled";
 
 const VIEW_MATERIALIZATION: &str = "view";
 const GRAPH_ARGUMENTS: [&str; 4] = ["selected", "upstream", "is_clonable", "is_view"];
 
 /// Whether a model is explicitly disabled with a boolean `enabled = false`.
 fn disabled_model(model: &Bound<'_, PyAny>) -> PyResult<bool> {
-    let raw = native_field(&native_field(model, "config")?, "values")?
-        .call_method1("get", ("enabled",))?;
-    Ok(raw.is_instance_of::<PyBool>() && !raw.is_truthy()?)
+    Ok(matches!(
+        *config_node_of(&model_values(model)?, ENABLED_KEY)?,
+        ConfigNode::Bool(false)
+    ))
 }
 
 fn tuple_of<'py>(py: Python<'py>, items: Vec<Bound<'py, PyAny>>) -> PyResult<Bound<'py, PyAny>> {

crates/sqlbuild-python/src/bindings/_helpers/planner_identity/metadata.rs

@@ -17,6 +17,7 @@ use sqlbuild_analysis::planner_identity::models::FunctionFingerprintFields;
 use crate::bindings::_helpers::boundary::python_calls::{imported, keywords, within};
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::discovery::hook_entries::hook_entries;
+use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
 use crate::bindings::_helpers::planner_identity::hashing::function_definition_hash;
 use crate::bindings::_helpers::planner_identity::json_values::{COMPACT, SPACED, sorted_dumps};
 use crate::bindings::_helpers::records::fields::field as native_field;
@@ -250,8 +251,7 @@ pub(crate) fn model_declared_columns_hash(model: &Bound<'_, PyAny>) -> PyResult<
     let payload: Bound<'_, PyDict> = PyDict::new(py);
     payload.set_item(
         "contract",
-        native_field(&native_field(model, "config")?, "values")?
-            .call_method1("get", (CONTRACT_CONFIG_KEY,))?,
+        config_value(py, &model_values(model)?, CONTRACT_CONFIG_KEY)?,
     )?;
     let columns: Bound<'_, PyList> = PyList::empty(py);
     let mut families: Bound<'_, PyList> = PyList::empty(py);

crates/sqlbuild-python/src/bindings/_helpers/planner_migrations/compatibility.rs

@@ -4,6 +4,7 @@ use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::compiled_views::model_reads::inferred_rows;
+use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
 use crate::bindings::_helpers::planner_changes::detect::detect_model_changes;
 use crate::bindings::_helpers::planner_migrations::models::{
     CompatibilityCheck, MigrationCompatibilityResult,
@@ -19,9 +20,7 @@ use sqlbuild_analysis::planner_migrations::main::compatibility_findings::compati
 use sqlbuild_analysis::planner_migrations::models::CompatibilityFinding;
 
 fn config(model: &Bound<'_, PyAny>, key: &str) -> PyResult<Option<String>> {
-    native_field(&native_field(model, "config")?, "values")?
-        .call_method1("get", (key,))?
-        .extract()
+    config_value(model.py(), &model_values(model)?, key)?.extract()
 }
 
 /// The schema findings that block migrating one model's history; `snapshots_config` is given for snapshots.

crates/sqlbuild-python/src/bindings/_helpers/planner_migrations/model_planning.rs

@@ -2,6 +2,9 @@
 
 use crate::bindings::_helpers::boundary::python_calls::call_with;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
+use crate::bindings::_helpers::model_config::typed_config::{
+    config_node_of, config_value, model_values,
+};
 use crate::bindings::_helpers::planner_changes::renames::plan_reference_renames;
 use crate::bindings::_helpers::planner_migrations::column_objects::{config, replace};
 use crate::bindings::_helpers::planner_migrations::discovery::discover_model_migrations;
@@ -16,18 +19,19 @@ use crate::bindings::_helpers::planner_phases::objects::object;
 use crate::bindings::_helpers::planner_phases::runtime::runtime_field;
 use crate::bindings::_helpers::records::construct::{enum_member, new_record};
 use crate::bindings::_helpers::records::fields::field as native_field;
-use crate::bindings::_helpers::records::fields::mapping_get;
 use crate::bindings::_helpers::records::fields::mapping_get_or;
 use crate::bindings::_helpers::records::ids::RecordId;
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyResult};
-use pyo3::types::{
-    PyBool, PyDict, PyDictMethods, PyFrozenSet, PySet, PySetMethods, PyString, PyTuple,
-};
+use pyo3::types::{PyDict, PyDictMethods, PyFrozenSet, PySet, PySetMethods, PyString, PyTuple};
+use sqlbuild_model_config::config_values::models::ConfigNode;
+
+const MIGRATE_FROM_KEY: &str = "migrate_from";
+const MIGRATE_FORCE_KEY: &str = "migrate_force";
 
 fn declared<'py>(item: &Bound<'py, PyAny>) -> PyResult<Option<ModelMigrationRequest<'py>>> {
     let py = item.py();
-    let values = native_field(&native_field(item, "config")?, "values")?;
-    let origin = mapping_get(&values, "migrate_from")?;
+    let values = model_values(item)?;
+    let origin = config_value(py, &values, MIGRATE_FROM_KEY)?;
     if !origin.is_instance_of::<PyString>() {
         return Ok(None);
     }
@@ -38,7 +42,10 @@ fn declared<'py>(item: &Bound<'py, PyAny>) -> PyResult<Option<ModelMigrationRequ
         raw_origin: Some(origin.call_method0("strip")?),
         origin_location: None,
         origin_model: py.None().into_bound(py),
-        force: mapping_get(&values, "migrate_force")?.is(PyBool::new(py, true)),
+        force: matches!(
+            *config_node_of(&values, MIGRATE_FORCE_KEY)?,
+            ConfigNode::Bool(true)
+        ),
         identity_only: materialization.eq("table")? || materialization.eq("view")?,
     }))
 }

crates/sqlbuild-python/src/bindings/_helpers/planner_phases/full_refresh.rs

@@ -1,6 +1,7 @@
 //! Per-model full-refresh resolution and full-rebuild refusals.
 
 use crate::bindings::_helpers::compiled_views::model_reads;
+use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
 use crate::bindings::_helpers::planner_phases::constants::FULL_REFRESH_DISABLED_REBUILD_CODE;
 use crate::bindings::_helpers::planner_phases::objects::{frozenset, planner_error, strings};
 use crate::bindings::_helpers::records::fields::field as native_field;
@@ -10,8 +11,10 @@ use pyo3::types::PyBool;
 use sqlbuild_analysis::planner_warehouse::main::full_rebuild_cause::full_rebuild_cause;
 use sqlbuild_analysis::planner_warehouse::models::FullRebuildFacts;
 
+const FULL_REFRESH_KEY: &str = "full_refresh";
+
 fn configured_full_refresh<'py>(model: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
-    native_field(&native_field(model, "config")?, "values")?.call_method1("get", ("full_refresh",))
+    config_value(model.py(), &model_values(model)?, FULL_REFRESH_KEY)
 }
 
 /// dbt-compatible nullable model full-refresh semantics.

crates/sqlbuild-python/src/bindings/_helpers/planner_phases/retention.rs

@@ -5,6 +5,7 @@ use sqlbuild_host::entries::models::AdapterHost;
 use crate::bindings::_helpers::boundary::python_calls::call_with;
 use crate::bindings::_helpers::compiled_views::model_reads;
 use crate::bindings::_helpers::contracts::spec_resolution::resolve_target_config;
+use crate::bindings::_helpers::model_config::typed_config::{config_value, typed_values};
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::planner_phases::constants::{
     BIGQUERY_ADAPTER, DECREASE_WARNING, DOWNGRADE_HISTORY_DAYS, PERMANENT_TABLE_TYPE,
@@ -28,6 +29,8 @@ use sqlbuild_analysis::planner_warehouse::main::retention_direction::retention_d
 use sqlbuild_analysis::planner_warehouse::main::table_type_copy_name::table_type_copy_name as copy_name;
 use std::collections::{BTreeMap, BTreeSet};
 
+const MATERIALIZED_KEY: &str = "materialized";
+
 /// The resolved config of the planned target, or defaults without one.
 #[pyfunction]
 #[pyo3(signature = (*, runtime))]
@@ -123,7 +126,7 @@ pub(crate) fn plan_table_types<'py>(
             &native_field(&materialization, "is_table_backed")?,
             &[(
                 "materialized",
-                native_field(&config, "values")?.call_method1("get", ("materialized",))?,
+                config_value(py, &typed_values(&config)?, MATERIALIZED_KEY)?,
             )],
         )?;
         if !backed.is_truthy()? {

crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/cursor_model.rs

@@ -6,8 +6,8 @@ use crate::bindings::_helpers::boundary::python_calls::call_with;
 use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::compile_attachment::cursor_roles::resolve_cursor_input_roles;
 use crate::bindings::_helpers::compiled_views::model_reads;
-use crate::bindings::_helpers::model_config::config_values::{
-    config_cursor_bound, config_str as authored_config_str,
+use crate::bindings::_helpers::model_config::typed_config::{
+    config_cursor_bound, config_text_object, model_values,
 };
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::plan_output::display::resolve_bounded_cursor_override;
@@ -35,13 +35,14 @@ use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyResult, Python};
 use pyo3::{IntoPyObject, PyErr};
 
 fn config<'py>(model: &Bound<'py, PyAny>, key: &str) -> PyResult<Bound<'py, PyAny>> {
-    let values: Bound<'py, PyAny> = native_field(&native_field(model, "config")?, "values")?;
+    let py: Python<'py> = model.py();
+    let values = model_values(model)?;
     let value: Option<Bound<'py, PyAny>> = if key == CURSOR_START {
-        config_cursor_bound(&values, key)?
+        config_cursor_bound(py, &values, key)?
     } else {
-        authored_config_str(&values, key)?
+        config_text_object(py, &values, key)?
     };
-    Ok(value.unwrap_or_else(|| model.py().None().into_bound(model.py())))
+    Ok(value.unwrap_or_else(|| py.None().into_bound(py)))
 }
 
 fn unbounded(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {

crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/cursor_overrides.rs

@@ -2,7 +2,9 @@
 
 use crate::bindings::_helpers::boundary::python_calls::{KeywordArguments, call_with};
 use crate::bindings::_helpers::compiled_views::model_reads;
-use crate::bindings::_helpers::model_config::config_values::config_str as authored_config_str;
+use crate::bindings::_helpers::model_config::typed_config::{
+    config_text_is, config_text_object, model_values,
+};
 use crate::bindings::_helpers::planner_resolution::values::{method, string};
 use crate::bindings::_helpers::records::construct::enum_from_value;
 use crate::bindings::_helpers::records::construct::error_class;
@@ -15,6 +17,13 @@ use pyo3::types::PyDict;
 use pyo3::{pyfunction, wrap_pyfunction};
 use sqlbuild_cursor::planner::main::override_kind::override_kind;
 
+const MATERIALIZED_KEY: &str = "materialized";
+const CURSOR_KEY: &str = "cursor";
+const CURSOR_TYPE_KEY: &str = "cursor_type";
+const INCREMENTAL: &str = "incremental";
+const INTEGER_CURSOR: &str = "integer";
+const TIMESTAMP_CURSOR: &str = "timestamp";
+
 #[pyfunction(name = "type_cursor_overrides")]
 #[pyo3(signature = (**kwargs))]
 fn type_cursor_overrides_of<'py>(
@@ -48,16 +57,15 @@ pub(crate) fn type_cursor_overrides<'py>(
     }
     let mut models: Vec<(String, String)> = Vec::new();
     for model in selected {
-        let values = native_field(&native_field(model, "config")?, "values")?;
-        let config = |key: &str| -> PyResult<Bound<'_, PyAny>> {
-            Ok(authored_config_str(&values, key)?.unwrap_or_else(|| py.None().into_bound(py)))
-        };
-        if !config("materialized")?.eq("incremental")? || config("cursor")?.is_none() {
+        let values = model_values(model)?;
+        let incremental: bool = config_text_is(py, &values, MATERIALIZED_KEY, INCREMENTAL)?;
+        if !incremental || config_text_object(py, &values, CURSOR_KEY)?.is_none() {
             continue;
         }
-        let kind = config("cursor_type")?;
-        if kind.eq("integer")? || kind.eq("timestamp")? {
-            models.push((model_reads::name_object(model)?.extract()?, kind.extract()?));
+        for kind in [INTEGER_CURSOR, TIMESTAMP_CURSOR] {
+            if config_text_is(py, &values, CURSOR_TYPE_KEY, kind)? {
+                models.push((model_reads::name_object(model)?.extract()?, kind.to_owned()));
+            }
         }
     }
     let mut integer_literals = true;

crates/sqlbuild-python/src/bindings/_helpers/planner_resolution/cursor_policies.rs

@@ -1,9 +1,9 @@
 //! Model-level cursor safety policy overrides over project configuration.
 
 use crate::bindings::_helpers::boundary::python_calls::call_with;
+use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values};
 use crate::bindings::_helpers::records::construct::{enum_from_value, record_class};
 use crate::bindings::_helpers::records::fields::field as native_field;
-use crate::bindings::_helpers::records::fields::mapping_get;
 use crate::bindings::_helpers::records::ids::RecordId;
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyModule, PyModuleMethods, PyResult};
 use pyo3::types::PyString;
@@ -45,9 +45,9 @@ fn resolve_policy<'py>(
     } else {
         class.call0()?
     };
-    let values = native_field(&native_field(model, "config")?, "values")?;
-    let raw_distance = mapping_get(&values, fields.distance_key)?;
-    let raw_action = mapping_get(&values, fields.action_key)?;
+    let values = model_values(model)?;
+    let raw_distance = config_value(py, &values, fields.distance_key)?;
+    let raw_action = config_value(py, &values, fields.action_key)?;
     let mut distance = configured.getattr(fields.distance)?;
     if raw_distance.is_instance_of::<PyString>() {
         distance = if raw_distance.eq(POLICY_DISABLED)? {