Support SQL default expressions in Python datasource schema

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "tinybird-sdk"
-version = "0.1.2"
+version = "0.1.3"
 description = "Python SDK for Tinybird Forward"
 readme = "README.md"
 authors = [

src/tinybird_sdk/generator/datasource.py

@@ -58,7 +58,10 @@ def _generate_column_line(column_name: str, column: TypeValidator | ColumnDefini
         parts.append(f"`json:{effective_json_path}`")
 
     if validator._modifiers.has_default:
-        parts.append(f"DEFAULT {_format_default_value(validator._modifiers.default_value, tinybird_type)}")
+        if isinstance(validator._modifiers.default_expression, str):
+            parts.append(f"DEFAULT {validator._modifiers.default_expression}")
+        else:
+            parts.append(f"DEFAULT {_format_default_value(validator._modifiers.default_value, tinybird_type)}")
 
     if validator._modifiers.codec:
         parts.append(f"CODEC({validator._modifiers.codec})")

src/tinybird_sdk/migrate/emit_ts.py

@@ -175,16 +175,19 @@ def _emit_datasource(ds: DatasourceModel) -> str:
         validator = _strict_column_type_to_validator(column.type)
 
         if column.default_expression is not None:
-            parsed_default = parse_literal_from_datafile(column.default_expression)
-            literal_value: Any = parsed_default
-            if isinstance(parsed_default, (int, float)) and _is_boolean_type(column.type):
-                if parsed_default in {0, 1}:
-                    literal_value = bool(parsed_default)
-                else:
-                    raise ValueError(
-                        f'Boolean default value must be 0 or 1 for column "{column.name}" in datasource "{ds.name}".'
-                    )
-            validator += f".default({repr(literal_value)})"
+            try:
+                parsed_default = parse_literal_from_datafile(column.default_expression)
+                literal_value: Any = parsed_default
+                if isinstance(parsed_default, (int, float)) and _is_boolean_type(column.type):
+                    if parsed_default in {0, 1}:
+                        literal_value = bool(parsed_default)
+                    else:
+                        raise ValueError(
+                            f'Boolean default value must be 0 or 1 for column "{column.name}" in datasource "{ds.name}".'
+                        )
+                validator += f".default({repr(literal_value)})"
+            except ValueError:
+                validator += f".default_expr({_escape_string(column.default_expression)})"
 
         if column.codec:
             validator += f".codec({_escape_string(column.codec)})"

src/tinybird_sdk/schema/types.py

@@ -12,6 +12,7 @@ class TypeModifiers:
     low_cardinality: bool = False
     has_default: bool = False
     default_value: Any = None
+    default_expression: str | None = None
     codec: str | None = None
 
 
@@ -53,7 +54,22 @@ def default(self, value: Any) -> "TypeValidator":
         return TypeValidator(
             _python_type=self._python_type,
             _tinybirdType=self._tinybirdType,
-            _modifiers=replace(self._modifiers, has_default=True, default_value=value),
+            _modifiers=replace(
+                self._modifiers, has_default=True, default_value=value, default_expression=None
+            ),
+        )
+
+    def default_expr(self, expression: str) -> "TypeValidator":
+        trimmed = expression.strip()
+        if not trimmed:
+            raise ValueError("Default expression cannot be empty.")
+
+        return TypeValidator(
+            _python_type=self._python_type,
+            _tinybirdType=self._tinybirdType,
+            _modifiers=replace(
+                self._modifiers, has_default=True, default_value=None, default_expression=trimmed
+            ),
         )
 
     def codec(self, codec: str) -> "TypeValidator":

tests/test_phase1_schema_generator_parity.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from tinybird_sdk import define_datasource, define_kafka_connection, t
+from tinybird_sdk import define_datasource, define_kafka_connection, get_modifiers, t
 from tinybird_sdk.generator.connection import generate_connection
 from tinybird_sdk.generator.datasource import generate_datasource
 
@@ -80,3 +80,31 @@ def test_generate_datasource_ignores_non_string_json_path(monkeypatch: pytest.Mo
 
     generated = generate_datasource(datasource)
     assert "`json:$.id`" in generated.content
+
+
+def test_type_validator_default_expr_stores_expression() -> None:
+    validator = t.uuid().default_expr("  generateUUIDv4()  ")
+    modifiers = get_modifiers(validator)
+
+    assert modifiers.has_default is True
+    assert modifiers.default_expression == "generateUUIDv4()"
+    assert modifiers.default_value is None
+
+
+def test_generate_datasource_emits_unquoted_default_expression() -> None:
+    datasource = define_datasource(
+        "events",
+        {
+            "schema": {
+                "id": t.uuid().default_expr("generateUUIDv4()"),
+            }
+        },
+    )
+
+    generated = generate_datasource(datasource)
+    assert "id UUID `json:$.id` DEFAULT generateUUIDv4()" in generated.content
+
+
+def test_type_validator_default_expr_rejects_empty_expression() -> None:
+    with pytest.raises(ValueError, match="Default expression cannot be empty."):
+        t.uuid().default_expr("   ")

tests/test_phase4_migrate_runner_emitter_parity.py

@@ -198,3 +198,33 @@ def test_run_migrate_rejects_sink_connection_type_mismatch(tmp_path: Path) -> No
 
     assert result.success is False
     assert any("is incompatible with connection" in error.message for error in result.errors)
+
+
+def test_run_migrate_emits_default_expr_for_sql_function_defaults(tmp_path: Path) -> None:
+    (tmp_path / "events.datasource").write_text(
+        "\n".join(
+            [
+                "SCHEMA >",
+                "    id UUID DEFAULT generateUUIDv4(),",
+                "    payload String DEFAULT '{}'",
+                "",
+                'ENGINE "MergeTree"',
+                'ENGINE_SORTING_KEY "id"',
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    result = run_migrate(
+        {
+            "cwd": str(tmp_path),
+            "patterns": ["*.datasource"],
+            "dry_run": True,
+        }
+    )
+
+    assert result.success is True
+    assert result.errors == []
+    assert result.output_content is not None
+    assert '\'id\': t.uuid().default_expr("generateUUIDv4()"),' in result.output_content
+    assert "'payload': t.string().default('{}')," in result.output_content