Ignore primary key, foreign key and unique constraints for DDL generation

Author: willmostly

tests/unit/sqlalchemy/test_compiler.py

@@ -18,6 +18,7 @@
 from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import Table
+from sqlalchemy import ForeignKey
 from sqlalchemy.schema import CreateTable
 from sqlalchemy.sql import column
 from sqlalchemy.sql import table
@@ -40,6 +41,26 @@
     trino_catalog='other'
 )
 
+table_with_pk = Table(
+    'table_with_pk',
+    metadata,
+    Column('id', String, primary_key=True)
+)
+
+table_with_fk = Table(
+    'table_with_fk',
+    metadata,
+    Column('id', String, primary_key=True),
+    Column('fk', String, ForeignKey('table_with_pk.id'))
+)
+
+table_with_unique = Table(
+    'table_with_constraint',
+    metadata,
+    Column('id', String, primary_key=True),
+    Column('uniq', String, unique=True)
+)
+
 
 @pytest.fixture
 def dialect():
@@ -170,3 +191,21 @@ def test_try_cast(dialect):
     statement = select(try_cast(table_without_catalog.c.id, String))
     query = statement.compile(dialect=dialect)
     assert str(query) == 'SELECT try_cast("table".id as VARCHAR) AS id \nFROM "table"'
+
+
+def test_catalogs_create_table_with_pk(dialect):
+    statement = CreateTable(table_with_pk)
+    query = statement.compile(dialect=dialect)
+    assert 'primary key' not in str(query).lower()
+
+
+def test_catalogs_create_table_with_fk(dialect):
+    statement = CreateTable(table_with_fk)
+    query = statement.compile(dialect=dialect)
+    assert 'foreign key' not in str(query).lower()
+
+
+def test_catalogs_create_table_with_unique(dialect):
+    statement = CreateTable(table_with_unique)
+    query = statement.compile(dialect=dialect)
+    assert 'unique' not in str(query).lower()

trino/sqlalchemy/compiler.py

@@ -181,7 +181,14 @@ def visit_try_cast(self, element, **kw):
 
 
 class TrinoDDLCompiler(compiler.DDLCompiler):
-    pass
+    def visit_foreign_key_constraint(self, constraint, **kw):
+        return None
+
+    def visit_primary_key_constraint(self, constraint, **kw):
+        return None
+
+    def visit_unique_constraint(self, constraint, **kw):
+        return None
 
 
 class TrinoTypeCompiler(compiler.GenericTypeCompiler):