Handle TextClause objects in DOMAIN expressions (#338)

andrew222651 · pre-commit-ci[bot] · agronholm · web-flow · commit 84dcc39d5cef · 2025-06-04T22:29:12.000+03:00
* Handle TextClause objects in DOMAIN expressions * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix line length * Update mypy config * Fix types * De-lint * Add test for integer type * Fix test * Handle TextClause objects in DOMAIN expressions * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix line length * Update mypy config * Fix types * De-lint * Add test for integer type * Fix test * Ignore no-untyped-call * Ignore no-untyped-call after merge * Convert all TextClause kwargs to text() * Move TextClause handling * Generalize text import * Update src/sqlacodegen/generators.py Co-authored-by: Alex Grönholm <alex.gronholm@nextday.fi> --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Alex Grönholm <alex.gronholm@nextday.fi>
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -28,7 +28,7 @@ repos:
       - id: mypy
         additional_dependencies:
           - pytest
-          - "sqlalchemy[mypy] < 2.0"
+          - "SQLAlchemy >= 2.0.29"
 
   - repo: https://github.com/pre-commit/pygrep-hooks
     rev: v1.10.0
diff --git a/pyproject.toml b/pyproject.toml
@@ -29,7 +29,7 @@ classifiers = [
 ]
 requires-python = ">=3.9"
 dependencies = [
-    "SQLAlchemy >= 2.0.23",
+    "SQLAlchemy >= 2.0.29",
     "inflect >= 4.0.0",
     "importlib_metadata; python_version < '3.10'",
 ]
@@ -80,6 +80,7 @@ extend-select = [
 
 [tool.mypy]
 strict = true
+disable_error_code = "no-untyped-call"
 
 [tool.pytest.ini_options]
 addopts = "-rsfE --tb=short"
diff --git a/src/sqlacodegen/generators.py b/src/sqlacodegen/generators.py
@@ -13,7 +13,7 @@
 from keyword import iskeyword
 from pprint import pformat
 from textwrap import indent
-from typing import Any, ClassVar
+from typing import Any, ClassVar, Literal, cast
 
 import inflect
 import sqlalchemy
@@ -38,7 +38,7 @@
     TypeDecorator,
     UniqueConstraint,
 )
-from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import DOMAIN, JSONB
 from sqlalchemy.engine import Connection, Engine
 from sqlalchemy.exc import CompileError
 from sqlalchemy.sql.elements import TextClause
@@ -228,6 +228,8 @@ def collect_imports_for_column(self, column: Column[Any]) -> None:
                 or column.type.astext_type.length is not None
             ):
                 self.add_import(column.type.astext_type)
+        elif isinstance(column.type, DOMAIN):
+            self.add_import(column.type.data_type.__class__)
 
         if column.default:
             self.add_import(column.default)
@@ -375,7 +377,7 @@ def render_table(self, table: Table) -> str:
 
             args.append(self.render_constraint(constraint))
 
-        for index in sorted(table.indexes, key=lambda i: i.name):
+        for index in sorted(table.indexes, key=lambda i: cast(str, i.name)):
             # One-column indexes should be rendered as index=True on columns
             if len(index.columns) > 1 or not uses_default_name(index):
                 args.append(self.render_index(index))
@@ -467,7 +469,7 @@ def render_column(
 
         if isinstance(column.server_default, DefaultClause):
             kwargs["server_default"] = render_callable(
-                "text", repr(column.server_default.arg.text)
+                "text", repr(cast(TextClause, column.server_default.arg).text)
             )
         elif isinstance(column.server_default, Computed):
             expression = str(column.server_default.sqltext)
@@ -514,12 +516,18 @@ def render_column_type(self, coltype: object) -> str:
 
             value = getattr(coltype, param.name, missing)
             default = defaults.get(param.name, missing)
+            if isinstance(value, TextClause):
+                self.add_literal_import("sqlalchemy", "text")
+                rendered_value = render_callable("text", repr(value.text))
+            else:
+                rendered_value = repr(value)
+
             if value is missing or value == default:
                 use_kwargs = True
             elif use_kwargs:
-                kwargs[param.name] = repr(value)
+                kwargs[param.name] = rendered_value
             else:
-                args.append(repr(value))
+                args.append(rendered_value)
 
         vararg = next(
             (
@@ -1072,6 +1080,7 @@ def generate_relationship_name(
                         preferred_name = column_names[0][:-3]
 
             if "use_inflect" in self.options:
+                inflected_name: str | Literal[False]
                 if relationship.type in (
                     RelationshipType.ONE_TO_MANY,
                     RelationshipType.MANY_TO_MANY,
@@ -1166,7 +1175,7 @@ def render_table_args(self, table: Table) -> str:
             args.append(self.render_constraint(constraint))
 
         # Render indexes
-        for index in sorted(table.indexes, key=lambda i: i.name):
+        for index in sorted(table.indexes, key=lambda i: cast(str, i.name)):
             if len(index.columns) > 1 or not uses_default_name(index):
                 args.append(self.render_index(index))
 
diff --git a/src/sqlacodegen/utils.py b/src/sqlacodegen/utils.py
@@ -2,7 +2,7 @@
 
 import re
 from collections.abc import Mapping
-from typing import Any
+from typing import Any, Literal, cast
 
 from sqlalchemy import PrimaryKeyConstraint, UniqueConstraint
 from sqlalchemy.engine import Connection, Engine
@@ -97,6 +97,7 @@ def uses_default_name(constraint: Constraint | Index) -> bool:
                 }
             )
 
+    key: Literal["fk", "pk", "ix", "ck", "uq"]
     if isinstance(constraint, Index):
         key = "ix"
     elif isinstance(constraint, CheckConstraint):
@@ -139,7 +140,10 @@ def uses_default_name(constraint: Constraint | Index) -> bool:
         raise TypeError(f"Unknown constraint type: {constraint.__class__.__qualname__}")
 
     try:
-        convention: str = table.metadata.naming_convention[key]
+        convention = cast(
+            Mapping[str, str],
+            table.metadata.naming_convention,
+        )[key]
         return constraint.name == (convention % values)
     except KeyError:
         return False
diff --git a/tests/test_generator_tables.py b/tests/test_generator_tables.py
@@ -205,6 +205,80 @@ def test_enum_detection(generator: CodeGenerator) -> None:
     )
 
 
+@pytest.mark.parametrize("engine", ["postgresql"], indirect=["engine"])
+def test_domain_text(generator: CodeGenerator) -> None:
+    Table(
+        "simple_items",
+        generator.metadata,
+        Column(
+            "postal_code",
+            postgresql.DOMAIN(
+                "us_postal_code",
+                Text,
+                constraint_name="valid_us_postal_code",
+                not_null=False,
+                check=text("VALUE ~ '^\\d{5}$' OR VALUE ~ '^\\d{5}-\\d{4}$'"),
+            ),
+            nullable=False,
+        ),
+    )
+
+    validate_code(
+        generator.generate(),
+        """\
+        from sqlalchemy import Column, MetaData, Table, Text, text
+        from sqlalchemy.dialects.postgresql import DOMAIN
+
+        metadata = MetaData()
+
+
+        t_simple_items = Table(
+            'simple_items', metadata,
+            Column('postal_code', DOMAIN('us_postal_code', Text(), \
+constraint_name='valid_us_postal_code', not_null=False, \
+check=text("VALUE ~ '^\\\\d{5}$' OR VALUE ~ '^\\\\d{5}-\\\\d{4}$'")), nullable=False)
+        )
+        """,
+    )
+
+
+@pytest.mark.parametrize("engine", ["postgresql"], indirect=["engine"])
+def test_domain_int(generator: CodeGenerator) -> None:
+    Table(
+        "simple_items",
+        generator.metadata,
+        Column(
+            "n",
+            postgresql.DOMAIN(
+                "positive_int",
+                INTEGER,
+                constraint_name="positive",
+                not_null=False,
+                check=text("VALUE > 0"),
+            ),
+            nullable=False,
+        ),
+    )
+
+    validate_code(
+        generator.generate(),
+        """\
+        from sqlalchemy import Column, INTEGER, MetaData, Table, text
+        from sqlalchemy.dialects.postgresql import DOMAIN
+
+        metadata = MetaData()
+
+
+        t_simple_items = Table(
+            'simple_items', metadata,
+            Column('n', DOMAIN('positive_int', INTEGER(), \
+constraint_name='positive', not_null=False, \
+check=text('VALUE > 0')), nullable=False)
+        )
+        """,
+    )
+
+
 @pytest.mark.parametrize("engine", ["postgresql"], indirect=["engine"])
 def test_column_adaptation(generator: CodeGenerator) -> None:
     Table(

Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,7 @@ classifiers = [`
`29`	`29`	`]`
`30`	`30`	`requires-python = ">=3.9"`
`31`	`31`	`dependencies = [`
`32`		`- "SQLAlchemy >= 2.0.23",`
	`32`	`+ "SQLAlchemy >= 2.0.29",`
`33`	`33`	`"inflect >= 4.0.0",`
`34`	`34`	`"importlib_metadata; python_version < '3.10'",`
`35`	`35`	`]`
`@@ -80,6 +80,7 @@ extend-select = [`
`80`	`80`
`81`	`81`	`[tool.mypy]`
`82`	`82`	`strict = true`
	`83`	`+disable_error_code = "no-untyped-call"`
`83`	`84`
`84`	`85`	`[tool.pytest.ini_options]`
`85`	`86`	`addopts = "-rsfE --tb=short"`