wrap models using macros

CHANGELOG.md

@@ -8,6 +8,8 @@
 - Ignore commented-out schema tests ([#330](https://github.com/fishtown-analytics/dbt/pull/330), [#328](https://github.com/fishtown-analytics/dbt/issues/328))
 - Fix run levels ([#343](https://github.com/fishtown-analytics/dbt/pull/343), [#340](https://github.com/fishtown-analytics/dbt/issues/340), [#338](https://github.com/fishtown-analytics/dbt/issues/338))
 - Fix concurrency, open a unique transaction per model ([#345](https://github.com/fishtown-analytics/dbt/pull/345), [#336](https://github.com/fishtown-analytics/dbt/issues/336))
+- Handle concurrent `DROP ... CASCADE`s in Redshift ([#349](https://github.com/fishtown-analytics/dbt/pull/349))
+- Always release connections (use `try .. finally`) ([#354](https://github.com/fishtown-analytics/dbt/pull/354))
 
 ### Changes
 

dbt/adapters/default.py

@@ -366,17 +366,17 @@ def release_connection(cls, profile, name):
     def cleanup_connections(cls):
         global connections_in_use, connections_available
 
-        for name, connection in connections_in_use.items():
-            if connection.get('state') != 'closed':
-                logger.debug("Connection '{}' was left open."
-                             .format(name))
-            else:
-                logger.debug("Connection '{}' was properly closed."
-                             .format(name))
-
         try:
             lock.acquire()
 
+            for name, connection in connections_in_use.items():
+                if connection.get('state') != 'closed':
+                    logger.debug("Connection '{}' was left open."
+                                 .format(name))
+                else:
+                    logger.debug("Connection '{}' was properly closed."
+                                 .format(name))
+
             # garbage collect, but don't close them in case someone
             # still has a handle
             connections_in_use = {}
@@ -525,3 +525,10 @@ def table_exists(cls, profile, schema, table, model_name=None):
         tables = cls.query_for_existing(profile, schema, model_name)
         exists = tables.get(table) is not None
         return exists
+
+    @classmethod
+    def already_exists(cls, profile, schema, table, model_name=None):
+        """
+        Alias for `table_exists`.
+        """
+        return cls.table_exists(profile, schema, table, model_name)

dbt/adapters/redshift.py

@@ -1,5 +1,11 @@
+import multiprocessing
+
 from dbt.adapters.postgres import PostgresAdapter
 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
+from dbt.compat import basestring
+
+
+drop_lock = multiprocessing.Lock()
 
 
 class RedshiftAdapter(PostgresAdapter):
@@ -30,7 +36,7 @@ def sort_qualifier(cls, sort_type, sort):
                 .format(sort_type, valid_sort_types)
             )
 
-        if type(sort) == str:
+        if isinstance(sort, basestring):
             sort_keys = [sort]
         else:
             sort_keys = sort
@@ -42,3 +48,28 @@ def sort_qualifier(cls, sort_type, sort):
         return "{sort_type} sortkey({keys_csv})".format(
             sort_type=sort_type, keys_csv=keys_csv
         )
+
+    @classmethod
+    def drop(cls, profile, relation, relation_type, model_name=None):
+        global drop_lock
+
+        to_return = None
+
+        try:
+            drop_lock.acquire()
+
+            connection = cls.get_connection(profile, model_name)
+
+            cls.commit(connection)
+            cls.begin(profile, connection.get('name'))
+
+            to_return = super(PostgresAdapter, cls).drop(
+                profile, relation, relation_type, model_name)
+
+            cls.commit(connection)
+            cls.begin(profile, connection.get('name'))
+
+            return to_return
+
+        finally:
+            drop_lock.release()