Fix problem with culvert geom migration in 228

CHANGES.rst

@@ -3,6 +3,25 @@ Changelog of threedi-schema
 
 
 
+0.229.1 (unreleased)
+--------------------
+
+- Fix issue where invalid geometries broke migration 228 for culverts
+
+
+0.229.0 (2025-01-08)
+--------------------
+
+- Rename sqlite table "tags" to "tag"
+- Remove indices referring to removed tables in previous migrations
+- Make model_settings.use_2d_rain and model_settings.friction_averaging booleans
+- Remove columns referencing v2 in geometry_column
+- Ensure correct use_* values when matching tables have no data
+- Use custom types for comma separated and table text fields to strip extra white space
+- Correct direction of dwf and surface map
+- Remove v2 related views from sqlite
+
+
 0.228.3 (2024-12-10)
 --------------------
 

threedi_schema/__init__.py

@@ -2,6 +2,6 @@
 from .domain import constants, custom_types, models  # NOQA
 
 # fmt: off
-__version__ = '0.228.3'
+__version__ = '0.229.1.dev0'
 
 # fmt: on

threedi_schema/domain/custom_types.py

@@ -1,6 +1,8 @@
+import re
+
 import geoalchemy2
 from packaging import version
-from sqlalchemy.types import Integer, TypeDecorator, VARCHAR
+from sqlalchemy.types import Integer, Text, TypeDecorator, VARCHAR
 
 
 class Geometry(geoalchemy2.types.Geometry):
@@ -66,6 +68,49 @@ class IntegerEnum(CustomEnum):
     impl = Integer
 
 
+def clean_csv_string(value: str) -> str:
+    return re.sub(r"\s*,\s*", ",", value.strip())
+
+
+class CSVText(TypeDecorator):
+    impl = Text
+    cache_ok = True
+
+    def process_bind_param(self, value, dialect):
+        if value is not None:
+            value = clean_csv_string(value)
+        return value
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            value = clean_csv_string(value)
+        return value
+
+
+def clean_csv_table(value: str) -> str:
+    # convert windows line endings to unix first
+    value = value.replace("\r\n", "\n")
+    # remove leading and trailing whitespace
+    value = value.strip()
+    # clean up each line
+    return "\n".join([clean_csv_string(line) for line in value.split("\n")])
+
+
+class CSVTable(TypeDecorator):
+    impl = Text
+    cache_ok = True
+
+    def process_bind_param(self, value, dialect):
+        if value is not None:
+            value = clean_csv_table(value)
+        return value
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            value = clean_csv_table(value)
+        return value
+
+
 class VarcharEnum(CustomEnum):
     cache_ok = True
     impl = VARCHAR