Add compatibility for JSON output of dbt>=0.19.0.

Summary:
Intended to resolve [issue dagster-io#3616](dagster-io#3616)

The JSON schema for dbt run results (among many other [dbt Artifacts](https://docs.getdbt.com/reference/artifacts/dbt-artifacts)) has [changed in dbt 0.19.0](https://github.com/fishtown-analytics/dbt/releases/tag/v0.19.0).

dagster-dbt currently fails when parsing the output from `dbt run` and `dbt compile`.  This diff sets missing fields as optional and should be compatible with dbt //before// and //after// 0.19.0

**To Do**
- [ ] Decide on how long dagster-dbt will support dbt <0.19.0. Please comment below with your thoughts.
- [ ] Include new metadata fields from dbt 0.19.0 in the dagster-dbt Outputs and AssetMaterializations

Test Plan: buildkite

Reviewers: sandyryza, max

Reviewed By: max

Differential Revision: https://dagster.phacility.com/D6407

python_modules/libraries/dagster-dbt/dagster_dbt/types.py

@@ -62,7 +62,7 @@ class NodeResult(
 
     def __new__(
         cls,
-        node: Dict[str, Any],
+        node: Optional[Dict[str, Any]] = None,
         error: Optional[str] = None,
         status: Optional[Union[str, int]] = None,
         execution_time: Optional[float] = None,
@@ -76,7 +76,7 @@ def __new__(
         step_timings = check.list_param(step_timings, "step_timings", of_type=StepTiming)
         return super().__new__(
             cls,
-            check.dict_param(node, "node", key_type=str),
+            check.opt_dict_param(node, "node", key_type=str),
             check.opt_str_param(error, "error"),
             status,
             check.opt_float_param(execution_time, "execution_time"),
@@ -99,7 +99,7 @@ def from_dict(cls, d: Dict[str, Any]) -> "NodeResult":
         Returns:
             NodeResult: An instance of :class:`NodeResult <dagster_dbt.NodeResult>`.
         """
-        node = check.dict_elem(d, "node")
+        node = check.opt_dict_elem(d, "node")
         error = check.opt_str_elem(d, "error")
         execution_time = check.float_elem(d, "execution_time")
         thread_id = check.opt_str_elem(d, "thread_id")
@@ -141,14 +141,14 @@ def __new__(
         cls,
         logs: List[Dict[str, Any]],
         results: List[NodeResult],
-        generated_at: str,
+        generated_at: Optional[str] = None,
         elapsed_time: Optional[float] = None,
     ):
         return super().__new__(
             cls,
             check.list_param(logs, "logs", of_type=Dict),
             results,
-            check.str_param(generated_at, "generated_at"),
+            check.opt_str_param(generated_at, "generated_at"),
             check.opt_float_param(elapsed_time, "elapsed_time"),
         )
 
@@ -167,7 +167,7 @@ def from_dict(cls, d: Dict[str, Any]) -> "DbtResult":
         logs = check.is_list(d["logs"], of_type=Dict)
         check.list_elem(d, "results")
         results = [NodeResult.from_dict(d) for d in check.is_list(d["results"], of_type=Dict)]
-        generated_at = check.str_elem(d, "generated_at")
+        generated_at = check.opt_str_elem(d, "generated_at")
         elapsed_time = check.float_elem(d, "elapsed_time")
 
         return cls(

python_modules/libraries/dagster-dbt/dagster_dbt/utils.py

@@ -14,7 +14,7 @@ def generate_materializations(
     asset_key_prefix = check.opt_list_param(asset_key_prefix, "asset_key_prefix", of_type=str)
 
     for node_result in dbt_output.result.results:
-        if node_result.node["resource_type"] in ["model", "snapshot"]:
+        if node_result.node.get("resource_type", None) in ["model", "snapshot"]:
             success = not node_result.fail and not node_result.skip and not node_result.error
             if success:
                 entries = [

python_modules/libraries/dagster-dbt/setup.py

@@ -29,6 +29,6 @@ def get_version() -> str:
         ],
         packages=find_packages(exclude=["test"]),
         install_requires=["dagster", "dagster-pandas", "pandas", "requests", "attrs"],
-        extras_require={"test": ["dbt==0.17.*"]},
+        extras_require={"test": ["dbt>=0.17.0"]},
         zip_safe=False,
     )