Output run hooks (#696)

core/dbt/logger.py

@@ -67,6 +67,10 @@ def notice(self, msg, *args, **kwargs):
 
 # provide this for the cache.
 CACHE_LOGGER = logging.getLogger('dbt.cache')
+# add a dummy handler to avoid `No handlers could be found for logger`
+nothing_handler = logging.StreamHandler()
+nothing_handler.setLevel(CRITICAL)
+CACHE_LOGGER.addHandler(nothing_handler)
 # provide this for RPC connection logging
 RPC_LOGGER = logging.getLogger('dbt.rpc')
 

core/dbt/task/run.py

@@ -1,27 +1,67 @@
 from __future__ import print_function
 
+import time
+
 from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.node_types import NodeType, RunHookType
 from dbt.node_runners import ModelRunner, RPCExecuteRunner
 
 import dbt.exceptions
 import dbt.flags
-import dbt.ui.printer
 from dbt.contracts.graph.parsed import Hook
 from dbt.hooks import get_hook_dict
+from dbt.ui.printer import \
+    print_hook_start_line, \
+    print_hook_end_line, \
+    print_timestamped_line, \
+    print_run_end_messages, \
+    get_counts
 
 from dbt.compilation import compile_node
 from dbt.task.compile import CompileTask, RemoteCompileTask
 from dbt.utils import get_nodes_by_tags
 
 
+class Timer(object):
+    def __init__(self):
+        self.start = None
+        self.end = None
+
+    @property
+    def elapsed(self):
+        if self.start is None or self.end is None:
+            return None
+        return self.end - self.start
+
+    def __enter__(self):
+        self.start = time.time()
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_tracebck):
+        self.end = time.time()
+
+
 class RunTask(CompileTask):
+    def __init__(self, args, config):
+        super(RunTask, self).__init__(args, config)
+        self.ran_hooks = []
+
     def raise_on_first_error(self):
         return False
 
     def populate_adapter_cache(self, adapter):
         adapter.set_relations_cache(self.manifest)
 
+    def get_hook_sql(self, adapter, hook, idx, num_hooks, extra_context):
+        compiled = compile_node(adapter, self.config, hook, self.manifest,
+                                extra_context)
+        statement = compiled.wrapped_sql
+        hook_index = hook.get('index', num_hooks)
+        hook_dict = get_hook_dict(statement, index=hook_index)
+        if dbt.flags.STRICT_MODE:
+            Hook(**hook_dict)
+        return hook_dict.get('sql', '')
+
     def run_hooks(self, adapter, hook_type, extra_context):
 
         nodes = self.manifest.nodes.values()
@@ -33,22 +73,35 @@ def run_hooks(self, adapter, hook_type, extra_context):
         # b/c psycopg2 automatically begins a transaction when a connection
         # is created.
         adapter.clear_transaction()
+        if not ordered_hooks:
+            return
+        num_hooks = len(ordered_hooks)
+
+        plural = 'hook' if num_hooks == 1 else 'hooks'
+        print_timestamped_line("")
+        print_timestamped_line(
+            'Running {} {} {}'.format(num_hooks, hook_type, plural)
+        )
 
-        for i, hook in enumerate(ordered_hooks):
-            compiled = compile_node(adapter, self.config, hook,
-                                    self.manifest, extra_context)
-            statement = compiled.wrapped_sql
+        for idx, hook in enumerate(ordered_hooks, start=1):
+            sql = self.get_hook_sql(adapter, hook, idx, num_hooks,
+                                    extra_context)
 
-            hook_index = hook.get('index', len(hooks))
-            hook_dict = get_hook_dict(statement, index=hook_index)
+            # convert all whitespace sequences to single spaces
+            hook_text = ' '.join(sql.split())
+            hook_text = '{}.{}.{}'.format(hook.package_name, hook_type, idx)
+            print_hook_start_line(hook_text, idx, num_hooks)
 
-            if dbt.flags.STRICT_MODE:
-                Hook(**hook_dict)
+            with Timer() as timer:
+                if len(sql.strip()) > 0:
+                    status, _ = adapter.execute(sql, auto_begin=False,
+                                                fetch=False)
+            self.ran_hooks.append(hook)
 
-            sql = hook_dict.get('sql', '')
+            print_hook_end_line(hook_text, status, idx, num_hooks,
+                                timer.elapsed)
 
-            if len(sql.strip()) > 0:
-                adapter.execute(sql, auto_begin=False, fetch=False)
+        print_timestamped_line("")
 
     def safe_run_hooks(self, adapter, hook_type, extra_context):
         try:
@@ -57,19 +110,18 @@ def safe_run_hooks(self, adapter, hook_type, extra_context):
             logger.info("Database error while running {}".format(hook_type))
             raise
 
-    @classmethod
-    def print_results_line(cls, results, execution_time):
-        nodes = [r.node for r in results]
-        stat_line = dbt.ui.printer.get_counts(nodes)
+    def print_results_line(self, results, execution_time):
+        nodes = [r.node for r in results] + self.ran_hooks
+        stat_line = get_counts(nodes)
 
         execution = ""
 
         if execution_time is not None:
             execution = " in {execution_time:0.2f}s".format(
                 execution_time=execution_time)
 
-        dbt.ui.printer.print_timestamped_line("")
-        dbt.ui.printer.print_timestamped_line(
+        print_timestamped_line("")
+        print_timestamped_line(
             "Finished running {stat_line}{execution}."
             .format(stat_line=stat_line, execution=execution))
 
@@ -107,7 +159,7 @@ def get_runner_type(self):
 
     def task_end_messages(self, results):
         if results:
-            dbt.ui.printer.print_run_end_messages(results)
+            print_run_end_messages(results)
 
 
 class RemoteRunTask(RemoteCompileTask, RunTask):

core/dbt/ui/printer.py

@@ -49,7 +49,8 @@ def print_timestamped_line(msg, use_color=None):
     logger.info("{} | {}".format(get_timestamp(), msg))
 
 
-def print_fancy_output_line(msg, status, index, total, execution_time=None):
+def print_fancy_output_line(msg, status, index, total, execution_time=None,
+                            truncate=False):
     if index is None or total is None:
         progress = ''
     else:
@@ -60,6 +61,8 @@ def print_fancy_output_line(msg, status, index, total, execution_time=None):
         message=msg)
 
     justified = prefix.ljust(80, ".")
+    if truncate and len(justified) > 77:
+        justified = justified[:77] + '...'
 
     if execution_time is None:
         status_time = ""
@@ -83,6 +86,8 @@ def get_counts(flat_nodes):
 
         if node.get('resource_type') == NodeType.Model:
             t = '{} {}'.format(get_materialization(node), t)
+        elif node.get('resource_type') == NodeType.Operation:
+            t = 'hook'
 
         counts[t] = counts.get(t, 0) + 1
 
@@ -97,6 +102,18 @@ def print_start_line(description, index, total):
     print_fancy_output_line(msg, 'RUN', index, total)
 
 
+def print_hook_start_line(statement, index, total):
+    msg = 'START hook: {}'.format(statement)
+    print_fancy_output_line(msg, 'RUN', index, total, truncate=True)
+
+
+def print_hook_end_line(statement, status, index, total, execution_time):
+    msg = 'OK hook: {}'.format(statement)
+    # hooks don't fail into this path, so always green
+    print_fancy_output_line(msg, green(status), index, total,
+                            execution_time=execution_time, truncate=True)
+
+
 def print_skip_line(model, schema, relation, index, num_models):
     msg = 'SKIP relation {}.{}'.format(schema, relation)
     print_fancy_output_line(msg, yellow('SKIP'), index, num_models)