diff --git a/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py b/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py index 62eda640..5e661ca5 100644 --- a/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py +++ b/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( diff --git a/test/test_suite.py b/test/test_suite.py index 5559cfad..75fc9a3c 100644 --- a/test/test_suite.py +++ b/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"