[view] Support CREATE OR REPLACE

sqlalchemy_utils/view.py

@@ -7,15 +7,19 @@
 
 
 class CreateView(DDLElement):
-    def __init__(self, name, selectable, materialized=False):
+    def __init__(self, name, selectable, materialized=False, replace=False):
+        if materialized and replace:
+            raise ValueError("Cannot use CREATE OR REPLACE with materialized views")
         self.name = name
         self.selectable = selectable
         self.materialized = materialized
+        self.replace = replace
 
 
 @compiler.compiles(CreateView)
 def compile_create_materialized_view(element, compiler, **kw):
-    return 'CREATE {}VIEW {} AS {}'.format(
+    return 'CREATE {}{}VIEW {} AS {}'.format(
+        'OR REPLACE ' if element.replace else '',
         'MATERIALIZED ' if element.materialized else '',
         compiler.dialect.identifier_preparer.quote(element.name),
         compiler.sql_compiler.process(element.selectable, literal_binds=True),
@@ -124,7 +128,8 @@ def create_view(
     name,
     selectable,
     metadata,
-    cascade_on_drop=True
+    cascade_on_drop=True,
+    replace=False,
 ):
     """ Create a view on a given metadata
 
@@ -164,7 +169,11 @@ def create_view(
         metadata=None
     )
 
-    sa.event.listen(metadata, 'after_create', CreateView(name, selectable))
+    sa.event.listen(
+        metadata,
+        'after_create',
+        CreateView(name, selectable, replace=replace),
+    )
 
     @sa.event.listens_for(metadata, 'after_create')
     def create_indexes(target, connection, **kw):

tests/test_views.py

@@ -7,6 +7,7 @@
     create_view,
     refresh_materialized_view
 )
+from sqlalchemy_utils.view import CreateView
 from sqlalchemy_utils.compat import _select_args
 
 
@@ -121,16 +122,18 @@ def life_cycle(
         engine,
         metadata,
         column,
-        cascade_on_drop
+        cascade_on_drop,
+        replace=False,
     ):
         __table__ = create_view(
             name='trivial_view',
             selectable=sa.select(*_select_args(column)),
             metadata=metadata,
-            cascade_on_drop=cascade_on_drop
+            cascade_on_drop=cascade_on_drop,
+            replace=replace,
         )
-        __table__.create(engine)
-        __table__.drop(engine)
+        metadata.create_all(engine)
+        metadata.drop_all(engine)
 
 
 class SupportsCascade(TrivialViewTestCases):
@@ -164,13 +167,67 @@ def test_life_cycle_no_cascade(
         self.life_cycle(engine, Base.metadata, User.id, cascade_on_drop=False)
 
 
+class SupportsReplace(TrivialViewTestCases):
+    def test_life_cycle_replace(
+        self,
+        connection,
+        engine,
+        Base,
+        User
+    ):
+        self.life_cycle(
+            engine,
+            Base.metadata,
+            User.id,
+            cascade_on_drop=False,
+            replace=True,
+        )
+
+    def test_life_cycle_replace_existing(
+        self,
+        connection,
+        engine,
+        Base,
+        User
+    ):
+        __table__ = create_view(
+            name='trivial_view',
+            selectable=sa.select(*_select_args(User.id)),
+            metadata=Base.metadata,
+        )
+        Base.metadata.create_all(engine)
+        view = CreateView(
+            name='trivial_view',
+            selectable=sa.select(*_select_args(User.id)),
+            replace=True,
+        )
+        connection.execute(view)
+        connection.commit()
+        Base.metadata.drop_all(engine)
+
+    def test_replace_materialized(
+        self,
+        connection,
+        engine,
+        Base,
+        User
+    ):
+        with pytest.raises(ValueError):
+            CreateView(
+                name='trivial_view',
+                selectable=sa.select(*_select_args(User.id)),
+                materialized=True,
+                replace=True,
+            )
+
+
 @pytest.mark.usefixtures('postgresql_dsn')
-class TestPostgresTrivialView(SupportsCascade, SupportsNoCascade):
+class TestPostgresTrivialView(SupportsCascade, SupportsNoCascade, SupportsReplace):
     pass
 
 
 @pytest.mark.usefixtures('mysql_dsn')
-class TestMySqlTrivialView(SupportsCascade, SupportsNoCascade):
+class TestMySqlTrivialView(SupportsCascade, SupportsNoCascade, SupportsReplace):
     pass