diff --git a/crates/sqltide-python/src/project_discovery.rs b/crates/sqltide-python/src/project_discovery.rs index 11f574ee..f26d2bd1 100644 --- a/crates/sqltide-python/src/project_discovery.rs +++ b/crates/sqltide-python/src/project_discovery.rs @@ -962,7 +962,7 @@ fn immutable_config_value<'py>( /// Freeze nested configuration values while preserving authored key order. #[pyfunction] -fn immutable_config_pairs<'py>( +pub(crate) fn immutable_config_pairs<'py>( py: Python<'py>, values: Bound<'py, PyAny>, ) -> PyResult> { diff --git a/crates/sqltide-python/src/shapes/discovery_models.rs b/crates/sqltide-python/src/shapes/discovery_models.rs new file mode 100644 index 00000000..605df845 --- /dev/null +++ b/crates/sqltide-python/src/shapes/discovery_models.rs @@ -0,0 +1,543 @@ +//! Authored project specification shapes and the discovery results built from them. + +use pyo3::exceptions::PyValueError; +use pyo3::prelude::*; +use pyo3::types::PyTuple; + +use super::{Coerce, Given, add, add_class, coerce, factory, raised, shapes}; +use crate::project_discovery::immutable_config_pairs; + +const CONFIG_PAIRS: Coerce = Coerce::With(config_pairs); +const READ_ONLY_MAPPING: Coerce = Coerce::Maybe(&Coerce::ReadOnlyMapping); +const ROW_ERROR_POLICY: Coerce = Coerce::Maybe(&Coerce::Enum("RowErrorPolicy")); +const BOUNDED_REPLAY_FALLBACK: Coerce = Coerce::Maybe(&Coerce::Enum("BoundedReplayFallback")); +const RETENTION_REFERENCE: Coerce = Coerce::Enum("KafkaRetentionReference"); +const MAYBE_RETENTION_REFERENCE: Coerce = Coerce::Maybe(&RETENTION_REFERENCE); +const MAYBE_RETENTION_ORIGIN: Coerce = Coerce::Maybe(&Coerce::Enum("KafkaRetentionOrigin")); +const MAYBE_PIPELINE_MODE: Coerce = Coerce::Maybe(&Coerce::Enum("PipelineMode")); +const SOURCE_KIND: Coerce = Coerce::Enum("SourceKind"); +const SOURCE_NAME_ORIGIN: Coerce = Coerce::Enum("SourceNameOrigin"); +const QUERY_OR_SQL_FILE: &str = "Exactly one of 'query' or 'sql_file' must be provided"; + +/// Configuration pairs frozen by `immutable_config_pairs`, in authored order. +fn config_pairs<'py>(value: &Bound<'py, PyAny>) -> PyResult> { + Ok(immutable_config_pairs(value.py(), value.clone())?.into_any()) +} + +/// The bound value at `index`. +fn at<'a, 'py>(values: &'a [Bound<'py, PyAny>], index: usize) -> PyResult<&'a Bound<'py, PyAny>> { + values + .get(index) + .ok_or_else(|| PyValueError::new_err("shape bound too few values")) +} + +/// Normalizes the values at the given indexes, in the order the dataclass's `__post_init__` did. +fn coerce_in_order(values: &mut [Bound<'_, PyAny>], rules: &[(usize, Coerce)]) -> PyResult<()> { + for (index, rule) in rules { + let slot = values + .get_mut(*index) + .ok_or_else(|| PyValueError::new_err("shape bound too few values"))?; + let normalized = coerce(slot.py(), *rule, slot)?; + *slot = normalized; + } + Ok(()) +} + +/// Exactly one of the query and the SQL file is given. +fn query_or_sql_file(query: &Bound<'_, PyAny>, sql_file: &Bound<'_, PyAny>) -> PyResult<()> { + if query.is_truthy()? == sql_file.is_truthy()? { + return Err(raised( + query.py(), + "ProjectSpecError", + QUERY_OR_SQL_FILE.to_owned(), + )?); + } + Ok(()) +} + +/// `TransformStep`: normalized policies and settings, one query source and an ORDER BY. +fn transform_step(values: &mut [Bound<'_, PyAny>]) -> PyResult<()> { + coerce_in_order( + values, + &[ + (21, ROW_ERROR_POLICY), + (3, Coerce::Tuple), + (12, READ_ONLY_MAPPING), + (14, Coerce::Enum("ReplayAnchorMode")), + (16, BOUNDED_REPLAY_FALLBACK), + ], + )?; + query_or_sql_file(at(values, 8)?, at(values, 9)?)?; + let order_by: &Bound<'_, PyAny> = at(values, 3)?; + if !order_by.is_truthy()? { + return Err(raised( + order_by.py(), + "ProjectSpecError", + "TransformStep must declare at least one ORDER BY expression".to_owned(), + )?); + } + Ok(()) +} + +/// `ViewStep`: one query source. +fn view_step(values: &mut [Bound<'_, PyAny>]) -> PyResult<()> { + query_or_sql_file(at(values, 1)?, at(values, 2)?) +} + +/// `Pipeline`: the row error policy, transforms, mode and replay fallback normalized in order. +fn pipeline(values: &mut [Bound<'_, PyAny>]) -> PyResult<()> { + coerce_in_order( + values, + &[ + (11, ROW_ERROR_POLICY), + (2, Coerce::Tuple), + (3, Coerce::Enum("PipelineMode")), + (5, BOUNDED_REPLAY_FALLBACK), + ], + ) +} + +shapes! { + module = "sqltide.compiler.discovery.models"; + + /// One project source file loaded exactly once during discovery. + DiscoveredProjectFile @no_repr { file_path, relative_path, contents } + + /// Unexpanded adapter-owned connection values retained without secret repr output. + RawConnectionConfig @no_repr { values(default = Given::EmptyTuple, coerce = CONFIG_PAIRS) } + + /// Effective target-resolved background audit scheduler configuration. + AuditSchedulerConfig { enabled(default = Given::Bool(false)) } + + /// Optional authored scheduler override at target scope. + AuditSchedulerOverride { enabled(default = Given::None) } + + /// Effective target-resolved sensor dispatcher configuration. + SensorAutomationConfig { + enabled(default = Given::Bool(false)), + tick_retention_days(default = Given::Int(0)), + maximum_event_age_seconds(default = Given::None), + } + + /// Optional authored sensor automation override at target scope. + SensorAutomationOverride { + enabled(default = Given::None), + tick_retention_days(default = Given::None), + maximum_event_age_seconds(default = Given::None), + } + + /// Committed absolute limits applied before a build mutates its target. + BuildConfig { max_pipelines(default = Given::None) } + + /// Committed finite ClickHouse table DROP ceiling for one target. + DestructionConfig { max_table_size_to_drop_bytes(default = Given::None) } + + /// Committed browser presentation settings. + UiConfig { timezone(default = Given::Str("UTC")) } + + /// One directional exception to pipeline-local model ownership. + AllowedCrossPipelineReference { upstream_pipeline, downstream_pipeline } + + /// Committed project dependency-boundary policy. + ProjectDependencies { + model_reference_scope(default = Given::Member("ModelReferenceScope", "PROJECT"), coerce = Coerce::Enum("ModelReferenceScope")), + allowed_cross_pipeline_references(default = Given::EmptyTuple, coerce = Coerce::Tuple), + } + + /// One committed named target before local resolution. + ProjectTarget @no_repr { + database(default = Given::None), + connection(default = Given::Factory(factory::)), + variables(default = Given::EmptyTuple, coerce = CONFIG_PAIRS), + audit_scheduler(default = Given::Factory(factory::)), + sensors(default = Given::Factory(factory::)), + build(default = Given::Factory(factory::)), + destruction(default = Given::Factory(factory::)), + production(default = Given::Bool(false)), + } + + /// One local named target definition or override. + LocalProjectTarget @no_repr { + database(default = Given::None), + connection(default = Given::Factory(factory::)), + variables(default = Given::EmptyTuple, coerce = CONFIG_PAIRS), + audit_scheduler(default = Given::Factory(factory::)), + sensors(default = Given::Factory(factory::)), + } + + /// One breaking or non-breaking replay policy choice. + ReplayOnChangeRule { + mode(coerce = Coerce::Enum("ReplayOnChangeMode")), + lookback_seconds(default = Given::None), + } + + /// Replay scope selected independently for breaking and non-breaking changes. + ReplayOnChangePolicy { breaking(default = Given::None), non_breaking(default = Given::None) } + + /// Authored freshness thresholds as `` durations. + SourceFreshnessPolicy { warn_after(default = Given::None), error_after(default = Given::None) } + + /// Optional project- or pipeline-level audit policy defaults. + AuditDefaults { + severity(default = Given::None), + cadence_seconds(default = Given::None), + warmup_seconds(default = Given::None), + } + + /// Project-wide virtual deployment comparison thresholds. + DeploymentReadinessDefaults { + maximum_lag_seconds(default = Given::Float(30.0)), + minimum_staged_row_ratio(default = Given::Float(0.5)), + } + + /// Project-wide defaults for managed Kafka sources. + KafkaSourceDefaults { + naming_macro(default = Given::None), + retention(default = Given::None), + retention_template(default = Given::None, compare = false, repr = false), + } + + /// Typed retention applied after a table model's output schema is known. + ModelRetentionPolicy { + duration_seconds, + timestamp_column, + cap_at_column(default = Given::None), + when_missing(default = Given::Member("RetentionMissingBehavior", "ERROR"), coerce = Coerce::Enum("RetentionMissingBehavior")), + } + + /// Typed retention over SQLTide's fixed managed Kafka replay schema. + KafkaRetentionPolicy { + duration_seconds, + timestamp(default = Given::Member("KafkaRetentionReference", "BROKER"), coerce = RETENTION_REFERENCE), + fallback(default = Given::Member("KafkaRetentionReference", "LANDED"), coerce = RETENTION_REFERENCE), + cap_at(default = Given::None, coerce = MAYBE_RETENTION_REFERENCE), + } + + /// Project- or pipeline-wide table-model defaults. + ModelDefaults { + retention(default = Given::None), + retention_template(default = Given::None, compare = false, repr = false), + on_row_error(default = Given::None, coerce = ROW_ERROR_POLICY), + } + + /// Project-wide defaults grouped by source kind. + SourceDefaults { kafka(default = Given::Factory(factory::)) } + + /// Committed project-wide authored defaults. + ProjectDefaults { + managed_source_ttl(default = Given::None), + model_ttl(default = Given::None), + kafka_broker_list(default = Given::None), + pipeline_mode(default = Given::Member("PipelineMode", "DIRECT"), coerce = Coerce::Enum("PipelineMode")), + replay_on_change(default = Given::None), + bounded_replay_fallback(default = Given::None), + run_presumed_failed_after_seconds(default = Given::Int(600)), + freshness(default = Given::None), + audits(default = Given::Factory(factory::)), + deployment_readiness(default = Given::Factory(factory::)), + sources(default = Given::Factory(factory::)), + models(default = Given::Factory(factory::)), + } + + /// Local overrides for project-wide pipeline defaults. + LocalProjectDefaults { pipeline_mode(default = Given::None, coerce = MAYBE_PIPELINE_MODE) } + + /// Project-wide pipeline validation and model relation naming. + ProjectNaming { + pipeline_prefix(default = Given::Str("pl__")), + pipeline_naming_macro(default = Given::None), + table_prefix(default = Given::Str("tbl__")), + view_prefix(default = Given::Str("view__")), + } + + /// Optional pipeline overrides for project model relation prefixes. + PipelineNaming { table_prefix(default = Given::None), view_prefix(default = Given::None) } + + /// Operator warning and exact confirmation required before a protected build. + PipelineProtection { warning, confirmation } + + /// Committed TOML project configuration before invocation resolution. + AuthoredProjectConfig @no_repr { + name, + adapter, + default_target, + connection, + variables(coerce = CONFIG_PAIRS), + targets, + defaults(default = Given::Factory(factory::)), + naming(default = Given::Factory(factory::)), + audit_scheduler(default = Given::Factory(factory::)), + sensors(default = Given::Factory(factory::)), + build(default = Given::Factory(factory::)), + ui(default = Given::Factory(factory::)), + dependencies(default = Given::Factory(factory::)), + } + + /// Optional local TOML overrides before invocation resolution. + LocalProjectConfig @no_repr { + target(default = Given::None), + adapter(default = Given::None), + defaults(default = Given::Factory(factory::)), + connection(default = Given::Factory(factory::)), + variables(default = Given::EmptyTuple, coerce = CONFIG_PAIRS), + targets(default = Given::EmptyTuple), + } + + /// Retained committed and optional local configuration inputs. + LoadedProjectConfiguration @no_repr { project, local, project_source, local_source } + + /// One immutable connection-lazy project configuration for an invocation. + EffectiveProjectConfiguration @no_repr { + name, + adapter, + target_name, + database, + connection, + variables(coerce = CONFIG_PAIRS), + defaults, + naming(default = Given::Factory(factory::)), + audit_scheduler(default = Given::Factory(factory::)), + sensors(default = Given::Factory(factory::)), + build(default = Given::Factory(factory::)), + destruction(default = Given::Factory(factory::)), + ui(default = Given::Factory(factory::)), + production_target(default = Given::Bool(false)), + dependencies(default = Given::Factory(factory::)), + } + + /// Kafka engine settings for a landing step. + KafkaSettings @no_repr { + broker_list, + topic, + consumer_group(default = Given::None), + format(default = Given::Str("JSONAsString")), + ttl(default = Given::None), + settings(default = Given::None, coerce = READ_ONLY_MAPPING), + retention(default = Given::None), + retention_origin(default = Given::None, coerce = MAYBE_RETENTION_ORIGIN), + } + + /// A standardized Kafka landing step. + KafkaLandingStep @no_repr { + name, + kafka, + replay_boundary(default = Given::None), + freshness(default = Given::None), + name_origin(default = Given::Member("SourceNameOrigin", "EXPLICIT"), coerce = SOURCE_NAME_ORIGIN), + naming_macro(default = Given::None), + naming_macro_fingerprint(default = Given::None), + } + + /// User-declared source boundary column mapping. + ReplayBoundaryColumns { + partition(default = Given::None), + offset(default = Given::None), + timestamp(default = Given::None), + landed_at(default = Given::None), + cursor(default = Given::None), + } + + /// User-declared replay boundary contract for an adopted source. + ReplayBoundary { mode(coerce = Coerce::Enum("ReplayBoundaryMode")), columns } + + /// A replay-driving adopted source table. + ExternalTableSourceStep { + name, + kind(coerce = SOURCE_KIND), + table_name, + replay_boundary, + freshness(default = Given::None), + name_origin(default = Given::Member("SourceNameOrigin", "EXPLICIT"), coerce = SOURCE_NAME_ORIGIN), + } + + /// A scheduled pull from an external Postgres table, carrying no replay lineage. + PostgresRefreshSourceStep { + name, + kind(coerce = SOURCE_KIND), + host, + database, + table, + user, + refresh, + port(default = Given::Int(5432)), + password_env(default = Given::None), + append(default = Given::Bool(true)), + freshness(default = Given::None), + name_origin(default = Given::Member("SourceNameOrigin", "EXPLICIT"), coerce = SOURCE_NAME_ORIGIN), + } + + /// An authored MODEL(...) column declaration. + ModelColumnSpec { + name, + description(default = Given::None), + audits(default = Given::EmptyTuple), + } + + /// Adapter query settings scoped by execution phase. + ExecutionSettings { replay(default = Given::None, coerce = READ_ONLY_MAPPING) } + + /// A transform from one logical upstream node into a managed table. + TransformStep { + name, + source, + engine, + order_by, + relation_name(default = Given::None), + description(default = Given::None), + columns(default = Given::EmptyTuple), + audits(default = Given::EmptyTuple), + query(default = Given::None), + sql_file(default = Given::None), + partition_by(default = Given::None), + ttl(default = Given::None), + settings(default = Given::None), + execution_settings(default = Given::Factory(factory::)), + replay_anchor(default = Given::Member("ReplayAnchorMode", "AUTO")), + replay_on_change(default = Given::None), + bounded_replay_fallback(default = Given::None), + source_file_path(default = Given::None), + source_line(default = Given::Int(1)), + source_column(default = Given::Int(1)), + retention(default = Given::None), + on_row_error(default = Given::None), + } + @post_init(transform_step) + + /// A query-only terminal model with arbitrary logical upstreams. + ViewStep { + name, + query(default = Given::None), + sql_file(default = Given::None), + relation_name(default = Given::None), + description(default = Given::None), + columns(default = Given::EmptyTuple), + audits(default = Given::EmptyTuple), + source_file_path(default = Given::None), + source_line(default = Given::Int(1)), + source_column(default = Given::Int(1)), + } + @post_init(view_step) + + /// Effective project values retained by the current compilation model. + Project { + replay_on_change(default = Given::None), + bounded_replay_fallback(default = Given::None, coerce = BOUNDED_REPLAY_FALLBACK), + model_ttl(default = Given::None), + default_database(default = Given::None), + adapter(default = Given::Str("clickhouse")), + naming(default = Given::Factory(factory::)), + audit_defaults(default = Given::Factory(factory::)), + audit_scheduler(default = Given::Factory(factory::)), + model_retention(default = Given::None), + dependencies(default = Given::Factory(factory::)), + model_on_row_error(default = Given::None), + } + + /// A single authored streaming pipeline. + Pipeline { + name, + source, + transforms(default = Given::Factory(factory::)), + mode(default = Given::Member("PipelineMode", "DIRECT")), + replay_on_change(default = Given::None), + bounded_replay_fallback(default = Given::None), + naming(default = Given::Factory(factory::)), + protection(default = Given::None), + audit_defaults(default = Given::Factory(factory::)), + execution_settings(default = Given::Factory(factory::)), + model_retention(default = Given::None), + model_on_row_error(default = Given::None), + } + @post_init(pipeline) + + /// A discovered pipeline plus the file it was loaded from. + LoadedPipeline { pipeline, file_path, project(default = Given::None) } + + /// Parsed project configuration with its retained authored source. + LoadedProject @no_repr { + project, + source_file, + configuration(default = Given::None), + effective_configuration(default = Given::None), + source_files(default = Given::EmptyTuple), + macro_files(default = Given::EmptyTuple), + macro_registry(default = Given::None), + } + + /// One retained standalone source declaration file and its parsed sources. + DiscoveredSourceFile @no_repr { source_file, sources } + + /// One direct pipeline directory and its optional retained configuration. + DiscoveredPipelineDirectory @no_repr { pipeline_dir, config_file(default = Given::None) } + + /// All raw project inputs captured before semantic compilation. + DiscoveredProjectInputs @no_repr { + project_dir, + loaded_project, + source_files, + pipeline_directories, + model_files, + test_files, + audit_files, + macro_files, + access_file(default = Given::None), + constant_files(default = Given::EmptyTuple), + enum_files(default = Given::EmptyTuple), + } +} + +pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add_class::(module)?; + add( + module, + "DiscoveryKafkaSettings", + module.py().get_type::(), + ) +} diff --git a/crates/sqltide-python/src/shapes/mod.rs b/crates/sqltide-python/src/shapes/mod.rs index 8e52325e..5c0f09d9 100644 --- a/crates/sqltide-python/src/shapes/mod.rs +++ b/crates/sqltide-python/src/shapes/mod.rs @@ -18,6 +18,7 @@ use pyo3::types::{ mod cli_service; mod compiler_analysis; mod compiler_project; +mod discovery_models; mod executor_ops; mod planner_execution; mod sensors_dev; @@ -86,6 +87,8 @@ pub(crate) type PostInit = for<'py> fn(&mut [Bound<'py, PyAny>]) -> PyResult<()> /// One shape: its class name, fields in declaration order and optional post-init hook. pub(crate) struct ShapeSpec { + /// The public module path, which with the class name identifies the shape. + pub(crate) module: &'static str, pub(crate) qualname: &'static str, pub(crate) fields: &'static [FieldSpec], pub(crate) post_init: Option, @@ -151,6 +154,7 @@ macro_rules! shapes { impl $name { pub(crate) const SPEC: $crate::shapes::ShapeSpec = $crate::shapes::ShapeSpec { + module: $module, qualname: stringify!($name), fields: &[$($crate::shapes::FieldSpec { name: $crate::shapes::unraw(stringify!($field)), @@ -394,8 +398,9 @@ pub(crate) fn bind<'py>( Ok(values) } -/// Each shape's plain default values by class name, built once so every instance and the -/// `dataclasses` field table share one default object, as a dataclass's defaults are shared. +/// Each shape's plain default values by module path and class name, built once so every +/// instance and the `dataclasses` field table share one default object, as a dataclass's +/// defaults are shared. static DEFAULTS: PyOnceLock> = PyOnceLock::new(); /// The shape's default values in field order, `None` where a field is required or has a @@ -404,7 +409,8 @@ fn shared_defaults<'py>(py: Python<'py>, spec: &ShapeSpec) -> PyResult = DEFAULTS .get_or_init(py, || PyDict::new(py).unbind()) .bind(py); - if let Some(found) = table.get_item(spec.qualname)? { + let key: (&str, &str) = (spec.module, spec.qualname); + if let Some(found) = table.get_item(key)? { return Ok(found.cast_into()?); } let mut values: Vec> = Vec::with_capacity(spec.fields.len()); @@ -415,7 +421,7 @@ fn shared_defaults<'py>(py: Python<'py>, spec: &ShapeSpec) -> PyResult = PyTuple::new(py, values)?; - table.set_item(spec.qualname, &defaults)?; + table.set_item(key, &defaults)?; Ok(defaults) } @@ -870,6 +876,7 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { executor_ops::register(module)?; cli_service::register(module)?; sensors_dev::register(module)?; + discovery_models::register(module)?; lazy_constants( module, &[ diff --git a/python-boundary.toml b/python-boundary.toml index ff6e0b9a..47217848 100644 --- a/python-boundary.toml +++ b/python-boundary.toml @@ -8,7 +8,7 @@ # still carries planning, validation, rendering or other movable logic counts as movable until # that logic moves to Rust. # Native failures must surface as errors; `handled_native_errors` lists the reviewed exceptions. -remaining_lines = 1774 +remaining_lines = 1080 [[stays]] path = "adapter/classes/adapter.py" diff --git a/scripts/python_boundary/constants.py b/scripts/python_boundary/constants.py index 7ecf6f60..2fbab0b5 100644 --- a/scripts/python_boundary/constants.py +++ b/scripts/python_boundary/constants.py @@ -20,4 +20,5 @@ NATIVE_ALIASES: dict[str, tuple[str, str]] = { "DiscoveryRefType": ("RefType", "compiler/discovery/types.py"), "DiscoverySqlRelationType": ("SqlRelationType", "compiler/discovery/types.py"), + "DiscoveryKafkaSettings": ("KafkaSettings", "compiler/discovery/models.py"), } diff --git a/src/sqltide/_native.pyi b/src/sqltide/_native.pyi index 62ae6a84..c6ee8913 100644 --- a/src/sqltide/_native.pyi +++ b/src/sqltide/_native.pyi @@ -77,36 +77,6 @@ from sqltide.adapter.types import ( ) from sqltide.adapters.clickhouse.classes.clickhouse_connection import ClickHouseConnection from sqltide.adapters.clickhouse.types import RawClickHouseClient -from sqltide.compiler.discovery.models import ( - AllowedCrossPipelineReference, - AuditDefaults, - DiscoveredPipelineDirectory, - DiscoveredProjectFile, - DiscoveredProjectInputs, - DiscoveredSourceFile, - EffectiveProjectConfiguration, - ExecutionSettings, - ExternalTableSourceStep, - KafkaLandingStep, - KafkaRetentionPolicy, - LoadedPipeline, - LoadedProject, - LoadedProjectConfiguration, - ModelRetentionPolicy, - Pipeline, - PipelineNaming, - PipelineProtection, - PostgresRefreshSourceStep, - Project, - ProjectNaming, - ReplayOnChangePolicy, - SourceFreshnessPolicy, - TransformStep, - ViewStep, -) -from sqltide.compiler.discovery.models import ( - KafkaSettings as _DiscoveryKafkaSettings, -) from sqltide.compiler.macros.models import MacroContext from sqltide.compiler.macros.types import MacroFunction from sqltide.dev_server.models import ( @@ -3284,6 +3254,445 @@ class BuildCancellationOutcome: warehouse_termination_confirmed: bool | None error_message: str | None = None +# Authored project specification and discovery shapes. + +@final +@dataclass(frozen=True, repr=False) +class DiscoveredProjectFile: + file_path: Path + relative_path: Path + contents: str + +@final +@dataclass(frozen=True, repr=False) +class RawConnectionConfig: + values: tuple[tuple[str, object], ...] = () + +@final +@dataclass(frozen=True) +class AuditSchedulerConfig: + enabled: bool = False + +@final +@dataclass(frozen=True) +class AuditSchedulerOverride: + enabled: bool | None = None + +@final +@dataclass(frozen=True) +class SensorAutomationConfig: + enabled: bool = False + tick_retention_days: int = 0 + maximum_event_age_seconds: int | None = None + +@final +@dataclass(frozen=True) +class SensorAutomationOverride: + enabled: bool | None = None + tick_retention_days: int | None = None + maximum_event_age_seconds: int | None = None + +@final +@dataclass(frozen=True) +class BuildConfig: + max_pipelines: int | None = None + +@final +@dataclass(frozen=True) +class DestructionConfig: + max_table_size_to_drop_bytes: int | None = None + +@final +@dataclass(frozen=True) +class UiConfig: + timezone: str = "UTC" + +@final +@dataclass(frozen=True) +class AllowedCrossPipelineReference: + upstream_pipeline: str + downstream_pipeline: str + +@final +@dataclass(frozen=True) +class ProjectDependencies: + model_reference_scope: ModelReferenceScope | str = ... + allowed_cross_pipeline_references: tuple[AllowedCrossPipelineReference, ...] = () + +@final +@dataclass(frozen=True, repr=False) +class ProjectTarget: + database: str | None = None + connection: RawConnectionConfig = ... + variables: tuple[tuple[str, object], ...] = () + audit_scheduler: AuditSchedulerOverride = ... + sensors: SensorAutomationOverride = ... + build: BuildConfig = ... + destruction: DestructionConfig = ... + production: bool = False + +@final +@dataclass(frozen=True, repr=False) +class LocalProjectTarget: + database: str | None = None + connection: RawConnectionConfig = ... + variables: tuple[tuple[str, object], ...] = () + audit_scheduler: AuditSchedulerOverride = ... + sensors: SensorAutomationOverride = ... + +@final +@dataclass(frozen=True) +class ReplayOnChangeRule: + mode: ReplayOnChangeMode | str + lookback_seconds: int | None = None + +@final +@dataclass(frozen=True) +class ReplayOnChangePolicy: + breaking: ReplayOnChangeRule | None = None + non_breaking: ReplayOnChangeRule | None = None + +@final +@dataclass(frozen=True) +class SourceFreshnessPolicy: + warn_after: str | None = None + error_after: str | None = None + +@final +@dataclass(frozen=True) +class AuditDefaults: + severity: str | None = None + cadence_seconds: int | None = None + warmup_seconds: int | None = None + +@final +@dataclass(frozen=True) +class DeploymentReadinessDefaults: + maximum_lag_seconds: float = 30.0 + minimum_staged_row_ratio: float = 0.5 + +@final +@dataclass(frozen=True) +class KafkaSourceDefaults: + naming_macro: str | None = None + retention: KafkaRetentionPolicy | Literal[False] | None = None + retention_template: object | None = field(default=None, repr=False, compare=False) + +@final +@dataclass(frozen=True) +class ModelRetentionPolicy: + duration_seconds: int + timestamp_column: str + cap_at_column: str | None = None + when_missing: RetentionMissingBehavior | str = ... + +@final +@dataclass(frozen=True) +class KafkaRetentionPolicy: + duration_seconds: int + timestamp: KafkaRetentionReference | str = ... + fallback: KafkaRetentionReference | str = ... + cap_at: KafkaRetentionReference | str | None = None + +@final +@dataclass(frozen=True) +class ModelDefaults: + retention: ModelRetentionPolicy | Literal[False] | None = None + retention_template: object | None = field(default=None, repr=False, compare=False) + on_row_error: RowErrorPolicy | str | None = None + +@final +@dataclass(frozen=True) +class SourceDefaults: + kafka: KafkaSourceDefaults = ... + +@final +@dataclass(frozen=True) +class ProjectDefaults: + managed_source_ttl: str | None = None + model_ttl: str | None = None + kafka_broker_list: str | None = None + pipeline_mode: PipelineMode | str = ... + replay_on_change: ReplayOnChangePolicy | None = None + bounded_replay_fallback: BoundedReplayFallback | None = None + run_presumed_failed_after_seconds: int = ... + freshness: SourceFreshnessPolicy | None = None + audits: AuditDefaults = ... + deployment_readiness: DeploymentReadinessDefaults = ... + sources: SourceDefaults = ... + models: ModelDefaults = ... + +@final +@dataclass(frozen=True) +class LocalProjectDefaults: + pipeline_mode: PipelineMode | str | None = None + +@final +@dataclass(frozen=True) +class ProjectNaming: + pipeline_prefix: str = ... + pipeline_naming_macro: str | None = None + table_prefix: str = ... + view_prefix: str = ... + +@final +@dataclass(frozen=True) +class PipelineNaming: + table_prefix: str | None = None + view_prefix: str | None = None + +@final +@dataclass(frozen=True) +class PipelineProtection: + warning: str + confirmation: str + +@final +@dataclass(frozen=True, repr=False) +class AuthoredProjectConfig: + name: str + adapter: str + default_target: str + connection: RawConnectionConfig + variables: tuple[tuple[str, object], ...] + targets: tuple[tuple[str, ProjectTarget], ...] + defaults: ProjectDefaults = ... + naming: ProjectNaming = ... + audit_scheduler: AuditSchedulerConfig = ... + sensors: SensorAutomationConfig = ... + build: BuildConfig = ... + ui: UiConfig = ... + dependencies: ProjectDependencies = ... + +@final +@dataclass(frozen=True, repr=False) +class LocalProjectConfig: + target: str | None = None + adapter: str | None = None + defaults: LocalProjectDefaults = ... + connection: RawConnectionConfig = ... + variables: tuple[tuple[str, object], ...] = () + targets: tuple[tuple[str, LocalProjectTarget], ...] = () + +@final +@dataclass(frozen=True, repr=False) +class LoadedProjectConfiguration: + project: AuthoredProjectConfig + local: LocalProjectConfig + project_source: DiscoveredProjectFile + local_source: DiscoveredProjectFile | None + +@final +@dataclass(frozen=True, repr=False) +class EffectiveProjectConfiguration: + name: str + adapter: str + target_name: str + database: str | None + connection: RawConnectionConfig + variables: tuple[tuple[str, object], ...] + defaults: ProjectDefaults + naming: ProjectNaming = ... + audit_scheduler: AuditSchedulerConfig = ... + sensors: SensorAutomationConfig = ... + build: BuildConfig = ... + destruction: DestructionConfig = ... + ui: UiConfig = ... + production_target: bool = False + dependencies: ProjectDependencies = ... + +@final +@dataclass(frozen=True, repr=False) +class DiscoveryKafkaSettings: + broker_list: str + topic: str + consumer_group: str | None = None + format: str = ... + ttl: str | None = None + settings: Mapping[str, str] | None = None + retention: KafkaRetentionPolicy | Literal[False] | None = None + retention_origin: KafkaRetentionOrigin | str | None = None + +@final +@dataclass(frozen=True, repr=False) +class KafkaLandingStep: + name: str + kafka: DiscoveryKafkaSettings + replay_boundary: ReplayBoundary | None = None + freshness: SourceFreshnessPolicy | None = None + name_origin: SourceNameOrigin | str = ... + naming_macro: str | None = None + naming_macro_fingerprint: str | None = None + +@final +@dataclass(frozen=True) +class ReplayBoundaryColumns: + partition: str | None = None + offset: str | None = None + timestamp: str | None = None + landed_at: str | None = None + cursor: str | None = None + +@final +@dataclass(frozen=True) +class ReplayBoundary: + mode: ReplayBoundaryMode | str + columns: ReplayBoundaryColumns + +@final +@dataclass(frozen=True) +class ExternalTableSourceStep: + name: str + kind: SourceKind | str + table_name: str + replay_boundary: ReplayBoundary + freshness: SourceFreshnessPolicy | None = None + name_origin: SourceNameOrigin | str = ... + +@final +@dataclass(frozen=True) +class PostgresRefreshSourceStep: + name: str + kind: SourceKind | str + host: str + database: str + table: str + user: str + refresh: str + port: int = ... + password_env: str | None = None + append: bool = True + freshness: SourceFreshnessPolicy | None = None + name_origin: SourceNameOrigin | str = ... + +@final +@dataclass(frozen=True) +class ModelColumnSpec: + name: str + description: str | None = None + audits: tuple[object, ...] = () + +@final +@dataclass(frozen=True) +class ExecutionSettings: + replay: Mapping[str, str] | None = None + +@final +@dataclass(frozen=True) +class TransformStep: + name: str + source: str + engine: str + order_by: Sequence[str] + relation_name: str | None = None + description: str | None = None + columns: tuple[ModelColumnSpec, ...] = () + audits: tuple[object, ...] = () + query: str | None = None + sql_file: str | None = None + partition_by: str | None = None + ttl: str | None = None + settings: Mapping[str, str] | None = None + execution_settings: ExecutionSettings = ... + replay_anchor: ReplayAnchorMode | str = ... + replay_on_change: ReplayOnChangePolicy | None = None + bounded_replay_fallback: BoundedReplayFallback | str | None = None + source_file_path: Path | None = None + source_line: int = 1 + source_column: int = 1 + retention: ModelRetentionPolicy | Literal[False] | None = None + on_row_error: RowErrorPolicy | str | None = None + +@final +@dataclass(frozen=True) +class ViewStep: + name: str + query: str | None = None + sql_file: str | None = None + relation_name: str | None = None + description: str | None = None + columns: tuple[ModelColumnSpec, ...] = () + audits: tuple[object, ...] = () + source_file_path: Path | None = None + source_line: int = 1 + source_column: int = 1 + +@final +@dataclass(frozen=True) +class Project: + replay_on_change: ReplayOnChangePolicy | None = None + bounded_replay_fallback: BoundedReplayFallback | str | None = None + model_ttl: str | None = None + default_database: str | None = None + adapter: str = ... + naming: ProjectNaming = ... + audit_defaults: AuditDefaults = ... + audit_scheduler: AuditSchedulerConfig = ... + model_retention: ModelRetentionPolicy | Literal[False] | None = None + dependencies: ProjectDependencies = ... + model_on_row_error: RowErrorPolicy | str | None = None + +@final +@dataclass(frozen=True) +class Pipeline: + name: str + source: KafkaLandingStep | ExternalTableSourceStep | PostgresRefreshSourceStep | None + transforms: Sequence[TransformStep | ViewStep] = ... + mode: PipelineMode | str = ... + replay_on_change: ReplayOnChangePolicy | None = None + bounded_replay_fallback: BoundedReplayFallback | str | None = None + naming: PipelineNaming = ... + protection: PipelineProtection | None = None + audit_defaults: AuditDefaults = ... + execution_settings: ExecutionSettings = ... + model_retention: ModelRetentionPolicy | Literal[False] | None = None + model_on_row_error: RowErrorPolicy | str | None = None + +@final +@dataclass(frozen=True) +class LoadedPipeline: + pipeline: Pipeline + file_path: Path + project: Project | None = None + +@final +@dataclass(frozen=True, repr=False) +class LoadedProject: + project: Project + source_file: DiscoveredProjectFile + configuration: LoadedProjectConfiguration | None = None + effective_configuration: EffectiveProjectConfiguration | None = None + source_files: tuple[DiscoveredSourceFile, ...] = () + macro_files: tuple[DiscoveredProjectFile, ...] = () + macro_registry: MacroRegistry | None = None + +@final +@dataclass(frozen=True, repr=False) +class DiscoveredSourceFile: + source_file: DiscoveredProjectFile + sources: tuple[KafkaLandingStep | ExternalTableSourceStep | PostgresRefreshSourceStep, ...] + +@final +@dataclass(frozen=True, repr=False) +class DiscoveredPipelineDirectory: + pipeline_dir: Path + config_file: DiscoveredProjectFile | None = None + +@final +@dataclass(frozen=True, repr=False) +class DiscoveredProjectInputs: + project_dir: Path + loaded_project: LoadedProject | None + source_files: tuple[DiscoveredSourceFile, ...] + pipeline_directories: tuple[DiscoveredPipelineDirectory, ...] + model_files: tuple[DiscoveredProjectFile, ...] + test_files: tuple[DiscoveredProjectFile, ...] + audit_files: tuple[DiscoveredProjectFile, ...] + macro_files: tuple[DiscoveredProjectFile, ...] + access_file: DiscoveredProjectFile | None = None + constant_files: tuple[DiscoveredProjectFile, ...] = () + enum_files: tuple[DiscoveredProjectFile, ...] = () + type _Span = tuple[int, int, int, int, int, int] class SqlScanError(ValueError): @@ -5513,16 +5922,16 @@ def ensure_header_columns( class KafkaLagSource(Protocol): """Cached consumer-group lag for one managed Kafka source.""" - def read(self, *, kafka: _DiscoveryKafkaSettings, database: str) -> KafkaLagSnapshot | None: ... + def read(self, *, kafka: DiscoveryKafkaSettings, database: str) -> KafkaLagSnapshot | None: ... class KafkaTopicSource(Protocol): """Cached topic inventory for one broker configuration.""" - def read(self, *, kafka: _DiscoveryKafkaSettings) -> KafkaTopicsSnapshot | None: ... + def read(self, *, kafka: DiscoveryKafkaSettings) -> KafkaTopicsSnapshot | None: ... -def build_kafka_client_config(*, kafka: _DiscoveryKafkaSettings) -> dict[str, object]: ... +def build_kafka_client_config(*, kafka: DiscoveryKafkaSettings) -> dict[str, object]: ... def kafka_client_key( - *, kafka: _DiscoveryKafkaSettings + *, kafka: DiscoveryKafkaSettings ) -> tuple[str, tuple[tuple[str, str], ...]]: ... def build_kafka_lag_snapshot( *, @@ -5542,7 +5951,7 @@ class KafkaLagReader: clock: Callable[[], float] | None = None, collector: Callable[..., object] | None = None, ) -> None: ... - def read(self, *, kafka: _DiscoveryKafkaSettings, database: str) -> KafkaLagSnapshot | None: ... + def read(self, *, kafka: DiscoveryKafkaSettings, database: str) -> KafkaLagSnapshot | None: ... def close(self) -> None: ... @final @@ -5555,7 +5964,7 @@ class KafkaTopicReader: clock: Callable[[], float] | None = None, collector: Callable[..., object] | None = None, ) -> None: ... - def read(self, *, kafka: _DiscoveryKafkaSettings) -> KafkaTopicsSnapshot | None: ... + def read(self, *, kafka: DiscoveryKafkaSettings) -> KafkaTopicsSnapshot | None: ... def close(self) -> None: ... @final diff --git a/src/sqltide/compiler/discovery/models.py b/src/sqltide/compiler/discovery/models.py index e8bdb261..0db2bedc 100644 --- a/src/sqltide/compiler/discovery/models.py +++ b/src/sqltide/compiler/discovery/models.py @@ -1,694 +1,51 @@ """Authored specification models and the discovery results built from them.""" -from __future__ import annotations - -from collections.abc import Mapping, Sequence -from dataclasses import dataclass, field -from pathlib import Path -from types import MappingProxyType -from typing import Literal - -import sqltide._native as _native -from sqltide.compiler.discovery.constants import ( - DEFAULT_ADAPTER_NAME, - DEFAULT_KAFKA_FORMAT, - DEFAULT_PIPELINE_PREFIX, - DEFAULT_POSTGRES_PORT, - DEFAULT_RUN_PRESUMED_FAILED_AFTER_SECONDS, - DEFAULT_TABLE_PREFIX, - DEFAULT_VIEW_PREFIX, -) -from sqltide.compiler.discovery.exceptions import ProjectSpecError -from sqltide.compiler.discovery.types import ( - BoundedReplayFallback, - KafkaRetentionOrigin, - KafkaRetentionReference, - ModelReferenceScope, - PipelineMode, - ReplayAnchorMode, - ReplayBoundaryMode, - ReplayOnChangeMode, - RetentionMissingBehavior, - RowErrorPolicy, - SourceKind, - SourceNameOrigin, -) -from sqltide.compiler.macros.models import MacroRegistry - - -@dataclass(frozen=True, repr=False) -class DiscoveredProjectFile: - """One project source file loaded exactly once during discovery.""" - - file_path: Path - relative_path: Path - contents: str - - -@dataclass(frozen=True, repr=False) -class RawConnectionConfig: - """Unexpanded adapter-owned connection values retained without secret repr output.""" - - values: tuple[tuple[str, object], ...] = () - - def __post_init__(self) -> None: - object.__setattr__(self, "values", _native.immutable_config_pairs(self.values)) - - -@dataclass(frozen=True) -class AuditSchedulerConfig: - """Effective target-resolved background audit scheduler configuration.""" - - enabled: bool = False - - -@dataclass(frozen=True) -class AuditSchedulerOverride: - """Optional authored scheduler override at target scope.""" - - enabled: bool | None = None - - -@dataclass(frozen=True) -class SensorAutomationConfig: - """Effective target-resolved sensor dispatcher configuration.""" - - enabled: bool = False - tick_retention_days: int = 0 - maximum_event_age_seconds: int | None = None - - -@dataclass(frozen=True) -class SensorAutomationOverride: - """Optional authored sensor automation override at target scope.""" - - enabled: bool | None = None - tick_retention_days: int | None = None - maximum_event_age_seconds: int | None = None - - -@dataclass(frozen=True) -class BuildConfig: - """Committed absolute limits applied before a build mutates its target.""" - - max_pipelines: int | None = None - - -@dataclass(frozen=True) -class DestructionConfig: - """Committed finite ClickHouse table DROP ceiling for one target.""" - - max_table_size_to_drop_bytes: int | None = None - - -@dataclass(frozen=True) -class UiConfig: - """Committed browser presentation settings.""" - - timezone: str = "UTC" - - -@dataclass(frozen=True) -class AllowedCrossPipelineReference: - """One directional exception to pipeline-local model ownership.""" - - upstream_pipeline: str - downstream_pipeline: str - - -@dataclass(frozen=True) -class ProjectDependencies: - """Committed project dependency-boundary policy.""" - - model_reference_scope: ModelReferenceScope | str = ModelReferenceScope.PROJECT - allowed_cross_pipeline_references: tuple[AllowedCrossPipelineReference, ...] = () - - def __post_init__(self) -> None: - object.__setattr__( - self, - "model_reference_scope", - ModelReferenceScope(self.model_reference_scope), - ) - object.__setattr__( - self, - "allowed_cross_pipeline_references", - tuple(self.allowed_cross_pipeline_references), - ) - - -@dataclass(frozen=True, repr=False) -class ProjectTarget: - """One committed named target before local resolution.""" - - database: str | None = None - connection: RawConnectionConfig = field(default_factory=RawConnectionConfig) - variables: tuple[tuple[str, object], ...] = () - audit_scheduler: AuditSchedulerOverride = field(default_factory=AuditSchedulerOverride) - sensors: SensorAutomationOverride = field(default_factory=SensorAutomationOverride) - build: BuildConfig = field(default_factory=BuildConfig) - destruction: DestructionConfig = field(default_factory=DestructionConfig) - production: bool = False - - def __post_init__(self) -> None: - object.__setattr__(self, "variables", _native.immutable_config_pairs(self.variables)) - - -@dataclass(frozen=True, repr=False) -class LocalProjectTarget: - """One local named target definition or override.""" - - database: str | None = None - connection: RawConnectionConfig = field(default_factory=RawConnectionConfig) - variables: tuple[tuple[str, object], ...] = () - audit_scheduler: AuditSchedulerOverride = field(default_factory=AuditSchedulerOverride) - sensors: SensorAutomationOverride = field(default_factory=SensorAutomationOverride) - - def __post_init__(self) -> None: - object.__setattr__(self, "variables", _native.immutable_config_pairs(self.variables)) - - -@dataclass(frozen=True) -class ReplayOnChangeRule: - """One breaking or non-breaking replay policy choice.""" - - mode: ReplayOnChangeMode | str - lookback_seconds: int | None = None - - def __post_init__(self) -> None: - object.__setattr__(self, "mode", ReplayOnChangeMode(self.mode)) - - -@dataclass(frozen=True) -class ReplayOnChangePolicy: - """Replay scope selected independently for breaking and non-breaking changes.""" - - breaking: ReplayOnChangeRule | None = None - non_breaking: ReplayOnChangeRule | None = None - - -@dataclass(frozen=True) -class SourceFreshnessPolicy: - """Authored freshness thresholds as `` durations.""" - - warn_after: str | None = None - error_after: str | None = None - - -@dataclass(frozen=True) -class AuditDefaults: - """Optional project- or pipeline-level audit policy defaults.""" - - severity: str | None = None - cadence_seconds: int | None = None - warmup_seconds: int | None = None - - -@dataclass(frozen=True) -class DeploymentReadinessDefaults: - """Project-wide virtual deployment comparison thresholds.""" - - maximum_lag_seconds: float = 30.0 - minimum_staged_row_ratio: float = 0.5 - - -@dataclass(frozen=True) -class KafkaSourceDefaults: - """Project-wide defaults for managed Kafka sources.""" - - naming_macro: str | None = None - retention: KafkaRetentionPolicy | Literal[False] | None = None - retention_template: object | None = field(default=None, repr=False, compare=False) - - -@dataclass(frozen=True) -class ModelRetentionPolicy: - """Typed retention applied after a table model's output schema is known.""" - - duration_seconds: int - timestamp_column: str - cap_at_column: str | None = None - when_missing: RetentionMissingBehavior | str = RetentionMissingBehavior.ERROR - - def __post_init__(self) -> None: - object.__setattr__(self, "when_missing", RetentionMissingBehavior(self.when_missing)) - - -@dataclass(frozen=True) -class KafkaRetentionPolicy: - """Typed retention over SQLTide's fixed managed Kafka replay schema.""" - - duration_seconds: int - timestamp: KafkaRetentionReference | str = KafkaRetentionReference.BROKER - fallback: KafkaRetentionReference | str = KafkaRetentionReference.LANDED - cap_at: KafkaRetentionReference | str | None = None - - def __post_init__(self) -> None: - object.__setattr__(self, "timestamp", KafkaRetentionReference(self.timestamp)) - object.__setattr__(self, "fallback", KafkaRetentionReference(self.fallback)) - if self.cap_at is not None: - object.__setattr__(self, "cap_at", KafkaRetentionReference(self.cap_at)) - - -@dataclass(frozen=True) -class ModelDefaults: - """Project- or pipeline-wide table-model defaults.""" - - retention: ModelRetentionPolicy | Literal[False] | None = None - retention_template: object | None = field(default=None, repr=False, compare=False) - on_row_error: RowErrorPolicy | str | None = None - - def __post_init__(self) -> None: - if self.on_row_error is not None: - object.__setattr__(self, "on_row_error", RowErrorPolicy(self.on_row_error)) - - -@dataclass(frozen=True) -class SourceDefaults: - """Project-wide defaults grouped by source kind.""" - - kafka: KafkaSourceDefaults = field(default_factory=KafkaSourceDefaults) - - -@dataclass(frozen=True) -class ProjectDefaults: - """Committed project-wide authored defaults.""" - - managed_source_ttl: str | None = None - model_ttl: str | None = None - kafka_broker_list: str | None = None - pipeline_mode: PipelineMode | str = PipelineMode.DIRECT - replay_on_change: ReplayOnChangePolicy | None = None - bounded_replay_fallback: BoundedReplayFallback | None = None - run_presumed_failed_after_seconds: int = DEFAULT_RUN_PRESUMED_FAILED_AFTER_SECONDS - freshness: SourceFreshnessPolicy | None = None - audits: AuditDefaults = field(default_factory=AuditDefaults) - deployment_readiness: DeploymentReadinessDefaults = field( - default_factory=DeploymentReadinessDefaults - ) - sources: SourceDefaults = field(default_factory=SourceDefaults) - models: ModelDefaults = field(default_factory=ModelDefaults) - - def __post_init__(self) -> None: - object.__setattr__(self, "pipeline_mode", PipelineMode(self.pipeline_mode)) - - -@dataclass(frozen=True) -class LocalProjectDefaults: - """Local overrides for project-wide pipeline defaults.""" - - pipeline_mode: PipelineMode | str | None = None - - def __post_init__(self) -> None: - if self.pipeline_mode is not None: - object.__setattr__(self, "pipeline_mode", PipelineMode(self.pipeline_mode)) - - -@dataclass(frozen=True) -class ProjectNaming: - """Project-wide pipeline validation and model relation naming.""" - - pipeline_prefix: str = DEFAULT_PIPELINE_PREFIX - pipeline_naming_macro: str | None = None - table_prefix: str = DEFAULT_TABLE_PREFIX - view_prefix: str = DEFAULT_VIEW_PREFIX - - -@dataclass(frozen=True) -class PipelineNaming: - """Optional pipeline overrides for project model relation prefixes.""" - - table_prefix: str | None = None - view_prefix: str | None = None - - -@dataclass(frozen=True) -class PipelineProtection: - """Operator warning and exact confirmation required before a protected build.""" - - warning: str - confirmation: str - - -@dataclass(frozen=True, repr=False) -class AuthoredProjectConfig: - """Committed TOML project configuration before invocation resolution.""" - - name: str - adapter: str - default_target: str - connection: RawConnectionConfig - variables: tuple[tuple[str, object], ...] - targets: tuple[tuple[str, ProjectTarget], ...] - defaults: ProjectDefaults = field(default_factory=ProjectDefaults) - naming: ProjectNaming = field(default_factory=ProjectNaming) - audit_scheduler: AuditSchedulerConfig = field(default_factory=AuditSchedulerConfig) - sensors: SensorAutomationConfig = field(default_factory=SensorAutomationConfig) - build: BuildConfig = field(default_factory=BuildConfig) - ui: UiConfig = field(default_factory=UiConfig) - dependencies: ProjectDependencies = field(default_factory=ProjectDependencies) - - def __post_init__(self) -> None: - object.__setattr__(self, "variables", _native.immutable_config_pairs(self.variables)) - - -@dataclass(frozen=True, repr=False) -class LocalProjectConfig: - """Optional local TOML overrides before invocation resolution.""" - - target: str | None = None - adapter: str | None = None - defaults: LocalProjectDefaults = field(default_factory=LocalProjectDefaults) - connection: RawConnectionConfig = field(default_factory=RawConnectionConfig) - variables: tuple[tuple[str, object], ...] = () - targets: tuple[tuple[str, LocalProjectTarget], ...] = () - - def __post_init__(self) -> None: - object.__setattr__(self, "variables", _native.immutable_config_pairs(self.variables)) - - -@dataclass(frozen=True, repr=False) -class LoadedProjectConfiguration: - """Retained committed and optional local configuration inputs.""" - - project: AuthoredProjectConfig - local: LocalProjectConfig - project_source: DiscoveredProjectFile - local_source: DiscoveredProjectFile | None - - -@dataclass(frozen=True, repr=False) -class EffectiveProjectConfiguration: - """One immutable connection-lazy project configuration for an invocation.""" - - name: str - adapter: str - target_name: str - database: str | None - connection: RawConnectionConfig - variables: tuple[tuple[str, object], ...] - defaults: ProjectDefaults - naming: ProjectNaming = field(default_factory=ProjectNaming) - audit_scheduler: AuditSchedulerConfig = field(default_factory=AuditSchedulerConfig) - sensors: SensorAutomationConfig = field(default_factory=SensorAutomationConfig) - build: BuildConfig = field(default_factory=BuildConfig) - destruction: DestructionConfig = field(default_factory=DestructionConfig) - ui: UiConfig = field(default_factory=UiConfig) - production_target: bool = False - dependencies: ProjectDependencies = field(default_factory=ProjectDependencies) - - def __post_init__(self) -> None: - object.__setattr__(self, "variables", _native.immutable_config_pairs(self.variables)) - - -@dataclass(frozen=True, repr=False) -class KafkaSettings: - """Kafka engine settings for a landing step.""" - - broker_list: str - topic: str - consumer_group: str | None = None - format: str = DEFAULT_KAFKA_FORMAT - ttl: str | None = None - settings: Mapping[str, str] | None = None - retention: KafkaRetentionPolicy | Literal[False] | None = None - retention_origin: KafkaRetentionOrigin | str | None = None - - def __post_init__(self) -> None: - if self.settings is not None: - object.__setattr__(self, "settings", MappingProxyType(dict(self.settings))) - if self.retention_origin is not None: - object.__setattr__( - self, "retention_origin", KafkaRetentionOrigin(self.retention_origin) - ) - - -@dataclass(frozen=True, repr=False) -class KafkaLandingStep: - """A standardized Kafka landing step.""" - - name: str - kafka: KafkaSettings - replay_boundary: ReplayBoundary | None = None - freshness: SourceFreshnessPolicy | None = None - name_origin: SourceNameOrigin | str = SourceNameOrigin.EXPLICIT - naming_macro: str | None = None - naming_macro_fingerprint: str | None = None - - def __post_init__(self) -> None: - object.__setattr__(self, "name_origin", SourceNameOrigin(self.name_origin)) - - -@dataclass(frozen=True) -class ReplayBoundaryColumns: - """User-declared source boundary column mapping.""" - - partition: str | None = None - offset: str | None = None - timestamp: str | None = None - landed_at: str | None = None - cursor: str | None = None - - -@dataclass(frozen=True) -class ReplayBoundary: - """User-declared replay boundary contract for an adopted source.""" - - mode: ReplayBoundaryMode | str - columns: ReplayBoundaryColumns - - def __post_init__(self) -> None: - object.__setattr__(self, "mode", ReplayBoundaryMode(self.mode)) - - -@dataclass(frozen=True) -class ExternalTableSourceStep: - """A replay-driving adopted source table.""" - - name: str - kind: SourceKind | str - table_name: str - replay_boundary: ReplayBoundary - freshness: SourceFreshnessPolicy | None = None - name_origin: SourceNameOrigin | str = SourceNameOrigin.EXPLICIT - - def __post_init__(self) -> None: - object.__setattr__(self, "kind", SourceKind(self.kind)) - object.__setattr__(self, "name_origin", SourceNameOrigin(self.name_origin)) - - -@dataclass(frozen=True) -class PostgresRefreshSourceStep: - """A scheduled pull from an external Postgres table, carrying no replay lineage.""" - - name: str - kind: SourceKind | str - host: str - database: str - table: str - user: str - refresh: str - port: int = DEFAULT_POSTGRES_PORT - password_env: str | None = None - append: bool = True - freshness: SourceFreshnessPolicy | None = None - name_origin: SourceNameOrigin | str = SourceNameOrigin.EXPLICIT - - def __post_init__(self) -> None: - object.__setattr__(self, "kind", SourceKind(self.kind)) - object.__setattr__(self, "name_origin", SourceNameOrigin(self.name_origin)) - - -@dataclass(frozen=True) -class ModelColumnSpec: - """An authored MODEL(...) column declaration.""" - - name: str - description: str | None = None - audits: tuple[object, ...] = () - - -@dataclass(frozen=True) -class ExecutionSettings: - """Adapter query settings scoped by execution phase.""" - - replay: Mapping[str, str] | None = None - - def __post_init__(self) -> None: - if self.replay is not None: - object.__setattr__(self, "replay", MappingProxyType(dict(self.replay))) - - -@dataclass(frozen=True) -class TransformStep: - """A transform from one logical upstream node into a managed table.""" - - name: str - source: str - engine: str - order_by: Sequence[str] - relation_name: str | None = None - description: str | None = None - columns: tuple[ModelColumnSpec, ...] = () - audits: tuple[object, ...] = () - query: str | None = None - sql_file: str | None = None - partition_by: str | None = None - ttl: str | None = None - settings: Mapping[str, str] | None = None - execution_settings: ExecutionSettings = field(default_factory=ExecutionSettings) - replay_anchor: ReplayAnchorMode | str = ReplayAnchorMode.AUTO - replay_on_change: ReplayOnChangePolicy | None = None - bounded_replay_fallback: BoundedReplayFallback | str | None = None - source_file_path: Path | None = None - source_line: int = 1 - source_column: int = 1 - retention: ModelRetentionPolicy | Literal[False] | None = None - on_row_error: RowErrorPolicy | str | None = None - - def __post_init__(self) -> None: - if self.on_row_error is not None: - object.__setattr__(self, "on_row_error", RowErrorPolicy(self.on_row_error)) - object.__setattr__(self, "order_by", tuple(self.order_by)) - if self.settings is not None: - object.__setattr__(self, "settings", MappingProxyType(dict(self.settings))) - object.__setattr__(self, "replay_anchor", ReplayAnchorMode(self.replay_anchor)) - if self.bounded_replay_fallback is not None: - object.__setattr__( - self, - "bounded_replay_fallback", - BoundedReplayFallback(self.bounded_replay_fallback), - ) - if bool(self.query) == bool(self.sql_file): - raise ProjectSpecError("Exactly one of 'query' or 'sql_file' must be provided") - if not self.order_by: - raise ProjectSpecError("TransformStep must declare at least one ORDER BY expression") - - -@dataclass(frozen=True) -class ViewStep: - """A query-only terminal model with arbitrary logical upstreams.""" - - name: str - query: str | None = None - sql_file: str | None = None - relation_name: str | None = None - description: str | None = None - columns: tuple[ModelColumnSpec, ...] = () - audits: tuple[object, ...] = () - source_file_path: Path | None = None - source_line: int = 1 - source_column: int = 1 - - def __post_init__(self) -> None: - if bool(self.query) == bool(self.sql_file): - raise ProjectSpecError("Exactly one of 'query' or 'sql_file' must be provided") - - -@dataclass(frozen=True) -class Project: - """Effective project values retained by the current compilation model.""" - - replay_on_change: ReplayOnChangePolicy | None = None - bounded_replay_fallback: BoundedReplayFallback | str | None = None - model_ttl: str | None = None - default_database: str | None = None - adapter: str = DEFAULT_ADAPTER_NAME - naming: ProjectNaming = field(default_factory=ProjectNaming) - audit_defaults: AuditDefaults = field(default_factory=AuditDefaults) - audit_scheduler: AuditSchedulerConfig = field(default_factory=AuditSchedulerConfig) - model_retention: ModelRetentionPolicy | Literal[False] | None = None - dependencies: ProjectDependencies = field(default_factory=ProjectDependencies) - model_on_row_error: RowErrorPolicy | str | None = None - - def __post_init__(self) -> None: - if self.bounded_replay_fallback is not None: - object.__setattr__( - self, - "bounded_replay_fallback", - BoundedReplayFallback(self.bounded_replay_fallback), - ) - - -@dataclass(frozen=True) -class Pipeline: - """A single authored streaming pipeline.""" - - name: str - source: KafkaLandingStep | ExternalTableSourceStep | PostgresRefreshSourceStep | None - transforms: Sequence[TransformStep | ViewStep] = field(default_factory=tuple) - mode: PipelineMode | str = PipelineMode.DIRECT - replay_on_change: ReplayOnChangePolicy | None = None - bounded_replay_fallback: BoundedReplayFallback | str | None = None - naming: PipelineNaming = field(default_factory=PipelineNaming) - protection: PipelineProtection | None = None - audit_defaults: AuditDefaults = field(default_factory=AuditDefaults) - execution_settings: ExecutionSettings = field(default_factory=ExecutionSettings) - model_retention: ModelRetentionPolicy | Literal[False] | None = None - model_on_row_error: RowErrorPolicy | str | None = None - - def __post_init__(self) -> None: - if self.model_on_row_error is not None: - object.__setattr__(self, "model_on_row_error", RowErrorPolicy(self.model_on_row_error)) - object.__setattr__(self, "transforms", tuple(self.transforms)) - object.__setattr__(self, "mode", PipelineMode(self.mode)) - if self.bounded_replay_fallback is not None: - object.__setattr__( - self, - "bounded_replay_fallback", - BoundedReplayFallback(self.bounded_replay_fallback), - ) - - -@dataclass(frozen=True) -class LoadedPipeline: - """A discovered pipeline plus the file it was loaded from.""" - - pipeline: Pipeline - file_path: Path - project: Project | None = None - - -@dataclass(frozen=True, repr=False) -class LoadedProject: - """Parsed project configuration with its retained authored source.""" - - project: Project - source_file: DiscoveredProjectFile - configuration: LoadedProjectConfiguration | None = None - effective_configuration: EffectiveProjectConfiguration | None = None - source_files: tuple[DiscoveredSourceFile, ...] = () - macro_files: tuple[DiscoveredProjectFile, ...] = () - macro_registry: MacroRegistry | None = None - - -@dataclass(frozen=True, repr=False) -class DiscoveredSourceFile: - """One retained standalone source declaration file and its parsed sources.""" - - source_file: DiscoveredProjectFile - sources: tuple[KafkaLandingStep | ExternalTableSourceStep | PostgresRefreshSourceStep, ...] - - -@dataclass(frozen=True, repr=False) -class DiscoveredPipelineDirectory: - """One direct pipeline directory and its optional retained configuration.""" - - pipeline_dir: Path - config_file: DiscoveredProjectFile | None = None - - -@dataclass(frozen=True, repr=False) -class DiscoveredProjectInputs: - """All raw project inputs captured before semantic compilation.""" - - project_dir: Path - loaded_project: LoadedProject | None - source_files: tuple[DiscoveredSourceFile, ...] - pipeline_directories: tuple[DiscoveredPipelineDirectory, ...] - model_files: tuple[DiscoveredProjectFile, ...] - test_files: tuple[DiscoveredProjectFile, ...] - audit_files: tuple[DiscoveredProjectFile, ...] - macro_files: tuple[DiscoveredProjectFile, ...] - access_file: DiscoveredProjectFile | None = None - constant_files: tuple[DiscoveredProjectFile, ...] = () - enum_files: tuple[DiscoveredProjectFile, ...] = () +from sqltide._native import AllowedCrossPipelineReference as AllowedCrossPipelineReference +from sqltide._native import AuditDefaults as AuditDefaults +from sqltide._native import AuditSchedulerConfig as AuditSchedulerConfig +from sqltide._native import AuditSchedulerOverride as AuditSchedulerOverride +from sqltide._native import AuthoredProjectConfig as AuthoredProjectConfig +from sqltide._native import BuildConfig as BuildConfig +from sqltide._native import DeploymentReadinessDefaults as DeploymentReadinessDefaults +from sqltide._native import DestructionConfig as DestructionConfig +from sqltide._native import DiscoveredPipelineDirectory as DiscoveredPipelineDirectory +from sqltide._native import DiscoveredProjectFile as DiscoveredProjectFile +from sqltide._native import DiscoveredProjectInputs as DiscoveredProjectInputs +from sqltide._native import DiscoveredSourceFile as DiscoveredSourceFile +from sqltide._native import DiscoveryKafkaSettings as KafkaSettings # noqa: F401 +from sqltide._native import EffectiveProjectConfiguration as EffectiveProjectConfiguration +from sqltide._native import ExecutionSettings as ExecutionSettings +from sqltide._native import ExternalTableSourceStep as ExternalTableSourceStep +from sqltide._native import KafkaLandingStep as KafkaLandingStep +from sqltide._native import KafkaRetentionPolicy as KafkaRetentionPolicy +from sqltide._native import KafkaSourceDefaults as KafkaSourceDefaults +from sqltide._native import LoadedPipeline as LoadedPipeline +from sqltide._native import LoadedProject as LoadedProject +from sqltide._native import LoadedProjectConfiguration as LoadedProjectConfiguration +from sqltide._native import LocalProjectConfig as LocalProjectConfig +from sqltide._native import LocalProjectDefaults as LocalProjectDefaults +from sqltide._native import LocalProjectTarget as LocalProjectTarget +from sqltide._native import ModelColumnSpec as ModelColumnSpec +from sqltide._native import ModelDefaults as ModelDefaults +from sqltide._native import ModelRetentionPolicy as ModelRetentionPolicy +from sqltide._native import Pipeline as Pipeline +from sqltide._native import PipelineNaming as PipelineNaming +from sqltide._native import PipelineProtection as PipelineProtection +from sqltide._native import PostgresRefreshSourceStep as PostgresRefreshSourceStep +from sqltide._native import Project as Project +from sqltide._native import ProjectDefaults as ProjectDefaults +from sqltide._native import ProjectDependencies as ProjectDependencies +from sqltide._native import ProjectNaming as ProjectNaming +from sqltide._native import ProjectTarget as ProjectTarget +from sqltide._native import RawConnectionConfig as RawConnectionConfig +from sqltide._native import ReplayBoundary as ReplayBoundary +from sqltide._native import ReplayBoundaryColumns as ReplayBoundaryColumns +from sqltide._native import ReplayOnChangePolicy as ReplayOnChangePolicy +from sqltide._native import ReplayOnChangeRule as ReplayOnChangeRule +from sqltide._native import SensorAutomationConfig as SensorAutomationConfig +from sqltide._native import SensorAutomationOverride as SensorAutomationOverride +from sqltide._native import SourceDefaults as SourceDefaults +from sqltide._native import SourceFreshnessPolicy as SourceFreshnessPolicy +from sqltide._native import TransformStep as TransformStep +from sqltide._native import UiConfig as UiConfig +from sqltide._native import ViewStep as ViewStep diff --git a/tests/unit/scripts/python_boundary/main/test_check.py b/tests/unit/scripts/python_boundary/main/test_check.py index e97597e3..0eb3d1e8 100644 --- a/tests/unit/scripts/python_boundary/main/test_check.py +++ b/tests/unit/scripts/python_boundary/main/test_check.py @@ -80,6 +80,18 @@ "PipelineMode as PipelineMode", "DiscoveryPipelineMode as PipelineMode" ) WRONG_ALIAS: str = DISCOVERY_TYPES.replace("DiscoveryRefType as RefType", "DiscoveryRefType as Ref") +DISCOVERY_MODELS: str = ( + '"""Discovery models, defined natively."""\n' + "\n" + "from sqltide._native import DiscoveryKafkaSettings as KafkaSettings\n" + "from sqltide._native import Project as Project\n" +) +UNLISTED_MODEL_ALIAS: str = DISCOVERY_MODELS.replace( + "Project as Project", "DiscoveryProject as Project" +) +WRONG_MODEL_ALIAS: str = DISCOVERY_MODELS.replace( + "DiscoveryKafkaSettings as KafkaSettings", "DiscoveryKafkaSettings as Settings" +) STAYS: str = '[[stays]]\npath = "executor"\nrule = 1\nnote = "Rule 1: runs plans."\n' REVIEWED: str = ( '[[handled_native_errors]]\nlocation = "compiler/keys.py::digest"\nnote = "No key."\n' @@ -103,6 +115,12 @@ policy="remaining_lines = 0\n", expected_messages=("1 pure native re-export files (4 lines)",), ), + PythonBoundaryTestCase( + description="the discovery Kafka settings alias in its own module is a pure re-export", + source_files={"compiler/discovery/models.py": DISCOVERY_MODELS}, + policy="remaining_lines = 0\n", + expected_messages=("1 pure native re-export files (4 lines)",), + ), PythonBoundaryTestCase( description="staying modules are not counted", source_files={"compiler/plan.py": "a = 1\nb = 2\n", "executor/run.py": "c = 3\n"}, @@ -208,6 +226,24 @@ def test_given_policy_matching_source_when_checking_then_passes( policy="remaining_lines = 0\n", expected_messages=("Movable Python grew to 4 lines (policy: 0)",), ), + PythonBoundaryTestCase( + description="an unlisted native alias beside the Kafka settings alias counts", + source_files={"compiler/discovery/models.py": UNLISTED_MODEL_ALIAS}, + policy="remaining_lines = 0\n", + expected_messages=("Movable Python grew to 4 lines (policy: 0)",), + ), + PythonBoundaryTestCase( + description="the discovery Kafka settings alias in the compile models counts", + source_files={"compiler/compile/models.py": DISCOVERY_MODELS}, + policy="remaining_lines = 0\n", + expected_messages=("Movable Python grew to 4 lines (policy: 0)",), + ), + PythonBoundaryTestCase( + description="the discovery Kafka settings under another alias counts", + source_files={"compiler/discovery/models.py": WRONG_MODEL_ALIAS}, + policy="remaining_lines = 0\n", + expected_messages=("Movable Python grew to 4 lines (policy: 0)",), + ), PythonBoundaryTestCase( description="a re-export module whose __all__ lists another name counts", source_files={"compiler/models.py": EXTRA_EXPORT}, diff --git a/tests/unit/src/sqltide/compiler/discovery/_test_types.py b/tests/unit/src/sqltide/compiler/discovery/_test_types.py index 9664f8d5..169811cd 100644 --- a/tests/unit/src/sqltide/compiler/discovery/_test_types.py +++ b/tests/unit/src/sqltide/compiler/discovery/_test_types.py @@ -1,7 +1,7 @@ from dataclasses import dataclass from pathlib import Path -from sqltide.compiler.discovery.models import SourceFreshnessPolicy +from sqltide.compiler.discovery.models import KafkaSettings, SourceFreshnessPolicy from sqltide.compiler.discovery.types import ModelReferenceScope, PipelineMode @@ -352,3 +352,11 @@ class ModelStepErrorTestCase: step_class: type arguments: dict[str, object] expected_message: str + + +@dataclass(frozen=True) +class KafkaSettingsReprTestCase: + description: str + kafka_settings: KafkaSettings + expected_repr_pattern: str + expected_hidden_values: tuple[str, ...] diff --git a/tests/unit/src/sqltide/compiler/discovery/test_models.py b/tests/unit/src/sqltide/compiler/discovery/test_models.py index c5bb3ace..23fd33e8 100644 --- a/tests/unit/src/sqltide/compiler/discovery/test_models.py +++ b/tests/unit/src/sqltide/compiler/discovery/test_models.py @@ -1,12 +1,20 @@ """Behavior tests for the authored model step specifications.""" +import re + import pytest from sqltide.compiler.discovery.exceptions import ProjectSpecError -from sqltide.compiler.discovery.models import TransformStep, ViewStep -from tests.unit.src.sqltide.compiler.discovery._test_types import ModelStepErrorTestCase +from sqltide.compiler.discovery.models import KafkaSettings, TransformStep, ViewStep +from tests.unit.src.sqltide.compiler.discovery._test_types import ( + KafkaSettingsReprTestCase, + ModelStepErrorTestCase, +) QUERY_OR_FILE_MESSAGE: str = "Exactly one of 'query' or 'sql_file' must be provided" +OBJECT_REPR_PATTERN: str = ( + r"" +) TRANSFORM_ARGUMENTS: dict[str, object] = { "name": "orders", "source": "raw_orders", @@ -68,3 +76,42 @@ def test_given_invalid_model_step_when_constructing_then_project_spec_error_is_r _ = test_case.step_class(**test_case.arguments) assert error_info.value.args == (test_case.expected_message,) + + +@pytest.mark.parametrize( + "test_case", + [ + KafkaSettingsReprTestCase( + description="landing Kafka settings keep the object repr and hide every field", + kafka_settings=KafkaSettings( + broker_list="broker-1.orders.example.com:9092", + topic="orders_topic", + consumer_group="orders_consumer_group", + format="JSONEachRow", + ttl="orders_ttl_expression", + settings={"kafka_sasl_password": "orders-secret-value"}, + retention_origin="project", + ), + expected_repr_pattern=OBJECT_REPR_PATTERN, + expected_hidden_values=( + "broker-1.orders.example.com", + "orders_topic", + "orders_consumer_group", + "JSONEachRow", + "orders_ttl_expression", + "kafka_sasl_password", + "orders-secret-value", + "KafkaRetentionOrigin", + "broker_list=", + ), + ), + ], + ids=lambda case: case.description, +) +def test_given_kafka_settings_when_rendering_repr_then_no_field_value_is_shown( + test_case: KafkaSettingsReprTestCase, +) -> None: + rendered: str = repr(test_case.kafka_settings) + + assert re.fullmatch(test_case.expected_repr_pattern, rendered) is not None + assert not any(value in rendered for value in test_case.expected_hidden_values)