add scaffolding

Signed-off-by: Massy Bourennani <[email protected]>

Author: MassyB

core/dbt/cli/requires.py

@@ -30,6 +30,7 @@
 from dbt.exceptions import DbtProjectError, FailFastError
 from dbt.flags import get_flag_dict, get_flags, set_flags
 from dbt.mp_context import get_mp_context
+from dbt.openlineage import OpenLineageHandler
 from dbt.parser.manifest import parse_manifest
 from dbt.plugins import set_up_plugin_manager
 from dbt.profiler import profiler
@@ -77,7 +78,9 @@ def wrapper(*args, **kwargs):
         reset_invocation_id()
 
         # Logging
-        callbacks = ctx.obj.get("callbacks", [])
+        ol_handler = OpenLineageHandler()
+        # todo replace here (conditional on OL config defined)
+        callbacks = ctx.obj.get("callbacks", [ol_handler.emit])
         setup_event_logger(flags=flags, callbacks=callbacks)
 
         # Tracking

core/dbt/events/core_types.proto

@@ -2262,3 +2262,14 @@ message ArtifactUploadSkippedMsg {
     CoreEventInfo info = 1;
     ArtifactUploadSkipped data = 2;
 }
+
+// Z064
+message OpenLineageException {
+    string exc = 1;
+    string exc_info = 2;
+}
+
+message OpenLineageExceptionMsg {
+    CoreEventInfo info = 1;
+    OpenLineageException data = 2;
+}

core/dbt/events/core_types_pb2.py

@@ -2212,3 +2212,16 @@ def code(self) -> str:
 
     def message(self) -> str:
         return f"Artifacts skipped for command : {self.msg}"
+
+
+# todo add a Test on this
+# https://github.com/dbt-labs/dbt-core/blob/main/core/dbt/events/README.md#adding-a-new-event
+class OpenLineageException(WarnLevel):
+    def code(self) -> str:
+        return "Z064"
+
+    def message(self):
+        return (
+            f"Encountered an error while creating OpenLineageEvent: {self.exc}\n"
+            f"{self.exc_info}"
+        )

core/dbt/events/types.py

@@ -0,0 +1 @@
+from .handler import OpenLineageHandler

core/dbt/openlineage/__init__.py

@@ -0,0 +1,31 @@
+import traceback
+
+from dbt.events.types import OpenLineageException
+from dbt_common.events.base_types import EventMsg
+from dbt_common.events.functions import fire_event
+
+
+class OpenLineageHandler:
+
+    def emit(self, e: EventMsg):
+        """
+        callback passed to the eventManager
+        """
+        if e.info.name == "OpenLineageException":
+            # not relevant to us
+            return
+        try:
+            self.emit_unsafe(e)
+        except Exception as e:
+            self._handle_exception(e)
+
+    def emit_unsafe(self, e: EventMsg):
+        # we only emit events for some family of structured logs not all of them
+        # import pdb; pdb.set_trace()
+
+        # todo add OL handling here
+
+        pass
+
+    def _handle_exception(self, e: Exception):
+        fire_event(OpenLineageException(exc=str(e), exc_info=traceback.format_exc()))