fix: NOT NULL computed column creation failure (#173)

Author: Ilya Gurov

google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -324,6 +324,29 @@ def limit_clause(self, select, **kw):
 class SpannerDDLCompiler(DDLCompiler):
     """Spanner DDL statements compiler."""
 
+    def get_column_specification(self, column, **kwargs):
+        """Build new column specifications.
+
+        Overridden to move the NOT NULL statement to front
+        of a computed column expression definitions.
+        """
+        colspec = (
+            self.preparer.format_column(column)
+            + " "
+            + self.dialect.type_compiler.process(column.type, type_expression=column)
+        )
+        default = self.get_column_default_string(column)
+        if default is not None:
+            colspec += " DEFAULT " + default
+
+        if not column.nullable:
+            colspec += " NOT NULL"
+
+        if column.computed is not None:
+            colspec += " " + self.process(column.computed)
+
+        return colspec
+
     def visit_computed_column(self, generated, **kw):
         """Computed column operator."""
         text = "AS (%s) STORED" % self.sql_compiler.process(

test/test_suite.py

@@ -1753,6 +1753,33 @@ def test_get_column_returns_computed(self):
         is_true("sqltext" in compData["computed"])
         eq_(self.normalize(compData["computed"]["sqltext"]), "normal+42")
 
+    def test_create_not_null_computed_column(self):
+        """
+        SPANNER TEST:
+
+        Check that on creating a computed column with a NOT NULL
+        clause the clause is set in front of the computed column
+        statement definition and doesn't cause failures.
+        """
+        engine = create_engine(get_db_url())
+        metadata = MetaData(bind=engine)
+
+        Table(
+            "Singers",
+            metadata,
+            Column("SingerId", String(36), primary_key=True, nullable=False),
+            Column("FirstName", String(200)),
+            Column("LastName", String(200), nullable=False),
+            Column(
+                "FullName",
+                String(400),
+                Computed("COALESCE(FirstName || ' ', '') || LastName"),
+                nullable=False,
+            ),
+        )
+
+        metadata.create_all(engine)
+
 
 @pytest.mark.skipif(
     bool(os.environ.get("SPANNER_EMULATOR_HOST")), reason="Skipped on emulator"