Address feedback, appease flake8/mypy/unit tests

core/dbt/contracts/results.py

@@ -12,15 +12,13 @@
 from hologram import JsonSchemaMixin
 
 import agate
-import logbook
 
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Union, Dict, List, Optional, Any, NamedTuple
 from numbers import Real
 
 
-
 @dataclass
 class TimingInfo(JsonSchemaMixin):
     name: str

core/dbt/logger.py

@@ -219,25 +219,18 @@ def process(self, record):
             record.extra['timing_info'] = self.timing_info.to_dict()
 
 
-class PushIfNonDefault(logbook.Processor):
-    KEY: str = NotImplemented
-    DEFAULT: Any = NotImplemented
+class DbtProcessState(logbook.Processor):
     def __init__(self, value: str):
         self.value = value
         super().__init__()
 
     def process(self, record):
         overwrite = (
-            self.KEY not in record.extra or
-            record.extra[self.KEY] == self.DEFAULT
+            'run_state' not in record.extra or
+            record.extra['run_state'] == 'internal'
         )
         if overwrite:
-            record.extra[self.KEY] = self.value
-
-
-class DbtProcessState(PushIfNonDefault):
-    KEY = 'run_state'
-    DEFAULT = 'internal'
+            record.extra['run_state'] = self.value
 
 
 class DbtModelState(logbook.Processor):
@@ -254,7 +247,7 @@ def process(self, record):
         record.extra['is_status_message'] = True
 
 
-class NodeInfo(logbook.Processor):
+class UniqueID(logbook.Processor):
     def __init__(self, unique_id: str):
         self.unique_id = unique_id
         super().__init__()
@@ -263,6 +256,29 @@ def process(self, record):
         record.extra['unique_id'] = self.unique_id
 
 
+class NodeMetadata(logbook.Processor):
+    def __init__(self, node):
+        self.node = node
+        super().__init__()
+
+    def process(self, record):
+        keys = [
+            ('alias', 'node_alias'),
+            ('schema', 'node_schema'),
+            ('database', 'node_database'),
+            ('name', 'node_name'),
+            ('original_file_path', 'node_path')
+        ]
+        for attr, key in keys:
+            value = getattr(self.node, attr, None)
+            if value is not None:
+                record.extra[key] = value
+        if hasattr(self.node, 'config'):
+            materialized = getattr(self.node.config, 'materialized', None)
+            if materialized is not None:
+                record.extra['node_materialized'] = materialized
+
+
 class TimestampNamed(JsonOnly):
     def __init__(self, name: str):
         self.name = name
@@ -273,7 +289,6 @@ def process(self, record):
         record.extra[self.name] = datetime.utcnow().isoformat()
 
 
-
 logger = logbook.Logger('dbt')
 # provide this for the cache, disabled by default
 CACHE_LOGGER = logbook.Logger('dbt.cache')
@@ -406,7 +421,7 @@ def __init__(self, stdout=colorama_stdout, stderr=sys.stderr):
         self._output_handler = OutputHandler(self.stdout)
         self._file_handler = DelayedFileHandler()
         self._relevel_processor = Relevel(allowed=['dbt', 'werkzeug'])
-        self._state_processor = DbtProcessState(DbtProcessState.DEFAULT)
+        self._state_processor = DbtProcessState('internal')
         # keep track of wheter we've already entered to decide if we should
         # be actually pushing. This allows us to log in main() and also
         # support entering dbt execution via handle_and_check.

core/dbt/node_runners.py

@@ -5,11 +5,7 @@
 
 from dbt import deprecations
 from dbt.adapters.base import BaseRelation
-from dbt.logger import (
-    GLOBAL_LOGGER as logger,
-    JsonOnly,
-    DbtModelState,
-)
+from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.exceptions import (
     NotImplementedException, CompilationException, RuntimeException,
     InternalException, missing_materialization

core/dbt/task/run.py

@@ -116,7 +116,6 @@ def run_hooks(self, adapter, hook_type, extra_context):
             print_hook_end_line(hook_text, status, idx, num_hooks,
                                 timer.elapsed)
 
-
         with TextOnly():
             print_timestamped_line("")
 

core/dbt/task/runnable.py

@@ -9,10 +9,10 @@
     GLOBAL_LOGGER as logger,
     DbtProcessState,
     TextOnly,
-    NodeInfo,
+    UniqueID,
     TimestampNamed,
-    JsonOnly,
     DbtModelState,
+    NodeMetadata,
 )
 from dbt.compilation import compile_manifest
 from dbt.contracts.results import ExecutionResult
@@ -127,8 +127,8 @@ def get_runner(self, node):
         return cls(self.config, adapter, node, run_count, num_nodes)
 
     def call_runner(self, runner):
-        with RUNNING_STATE, NodeInfo(runner.node.unique_id):
-            with TimestampNamed('node_started_at'):
+        with RUNNING_STATE, UniqueID(runner.node.unique_id):
+            with TimestampNamed('node_started_at'), NodeMetadata(runner.node):
                 logger.info('Began running model')
             status = 'error'  # we must have an error if we don't see this
             try: