Feature/52 auto dbt artifacts

dbterd/adapters/base.py

@@ -22,6 +22,7 @@ def __init__(self, ctx) -> None:
         self.ctx = ctx
         self.filename_manifest = "manifest.json"
         self.filename_catalog = "catalog.json"
+        self.dbt: DbtInvocation = None
 
     def run(self, **kwargs):
         """Main function helps to run by the target strategy"""
@@ -31,21 +32,31 @@ def run(self, **kwargs):
     def evaluate_kwargs(self, **kwargs) -> dict:
         """Re-calculate the options
 
+        - trigger `dbt ls` for re-calculate the Selection if `--dbt` enabled
+        - trigger `dbt docs generate` for re-calculate the artifact direction if `--dbt-atu-artifacts` enabled
+
         Raises:
             click.UsageError: Not Supported exception
 
         Returns:
             dict: kwargs dict
         """
         artifacts_dir, dbt_project_dir = self.__get_dir(**kwargs)
-        logger.info(f"Using dbt artifact dir at: {artifacts_dir}")
-        logger.info(f"Using dbt project  dir at: {dbt_project_dir}")
+        logger.info(f"Using dbt project dir at: {dbt_project_dir}")
 
         select = list(kwargs.get("select")) or []
         exclude = list(kwargs.get("exclude")) or []
         if kwargs.get("dbt"):
+            self.dbt = DbtInvocation(
+                dbt_project_dir=kwargs.get("dbt_project_dir"),
+                dbt_target=kwargs.get("dbt_target"),
+            )
             select = self.__get_selection(**kwargs)
             exclude = []
+
+            if kwargs.get("dbt_auto_artifacts"):
+                self.dbt.get_artifacts_for_erd()
+                artifacts_dir = f"{dbt_project_dir}/target"
         else:
             unsupported, rule = has_unsupported_rule(
                 rules=select.extend(exclude) if exclude else select
@@ -55,6 +66,7 @@ def evaluate_kwargs(self, **kwargs) -> dict:
                 logger.error(message)
                 raise click.UsageError(message)
 
+        logger.info(f"Using dbt artifact dir at: {artifacts_dir}")
         kwargs["artifacts_dir"] = artifacts_dir
         kwargs["dbt_project_dir"] = dbt_project_dir
         kwargs["select"] = select
@@ -91,10 +103,10 @@ def __get_dir(self, **kwargs) -> str:
 
     def __get_selection(self, **kwargs):
         """Override the Selection using dbt's one with `--dbt`"""
-        return DbtInvocation(
-            dbt_project_dir=kwargs.get("dbt_project_dir"),
-            dbt_target=kwargs.get("dbt_target"),
-        ).get_selection(
+        if not self.dbt:
+            raise click.UsageError("Flag `--dbt` need to be enabled")
+
+        return self.dbt.get_selection(
             select_rules=kwargs.get("select"),
             exclude_rules=kwargs.get("exclude"),
         )

dbterd/adapters/dbt_invocation.py

@@ -26,8 +26,52 @@ def __init__(self, dbt_project_dir: str = None, dbt_target: str = None) -> None:
             dbt_project_dir or os.environ.get("DBT_PROJECT_DIR") or str(Path.cwd())
         )
         self.target = dbt_target
+        self.args = ["--quiet", "--log-level", "none"]
+
+    def __invoke(self, runner_args: List[str] = []):
+        """Base function of the dbt invocation
+
+        Args:
+            runner_args (List[str], optional): Actual dbt arguments. Defaults to [].
+
+        Raises:
+            click.UsageError: Invocation failed for a reason
+
+        Returns:
+            dbtRunnerResult.result: dbtRunnerResult.result
+        """
+        args = self.__construct_arguments(*runner_args)
+        logger.debug(f"Invoking: `dbt {' '.join(args)}` at {self.project_dir}")
+        r: dbtRunnerResult = self.dbt.invoke(args)
+
+        if not r.success:
+            logger.error(str(r))
+            raise click.UsageError(str(r))
+
+        return r.result
+
+    def __construct_arguments(self, *args) -> List[str]:
+        """Enrich the dbt arguements with the based options
+
+        Returns:
+            List[str]: Actual dbt arguments
+        """
+        evaluated_args = args
+        if self.args:
+            evaluated_args = [*self.args, *args]
+        if self.project_dir:
+            evaluated_args.extend(["--project-dir", self.project_dir])
+        if self.target:
+            evaluated_args.extend(["--target", self.target])
+
+        return evaluated_args
 
     def __ensure_dbt_installed(self):
+        """Verify if dbt get installed
+
+        Raises:
+            click.UsageError: dbt is not installed
+        """
         dbt_spec = importlib.util.find_spec("dbt")
         if dbt_spec and dbt_spec.loader:
             installed_path = dbt_spec.submodule_search_locations[0]
@@ -55,22 +99,22 @@ def get_selection(
         Returns:
             List[str]: Selected node names with 'exact' rule
         """
-        args = ["ls", "--project-dir", self.project_dir, "--resource-type", "model"]
+        args = ["ls", "--resource-type", "model"]
         if select_rules:
             args.extend(["--select", " ".join(select_rules)])
         if exclude_rules:
             args.extend(["--exclude", " ".join(exclude_rules)])
-        if self.target:
-            args.extend(["--target", self.target])
-
-        logger.info(f"Invoking: `dbt {' '.join(args)}` at {self.project_dir}")
-        r: dbtRunnerResult = self.dbt.invoke(args)
-
-        if not r.success:
-            logger.error(str(r))
-            raise click.UsageError("str(r)")
 
+        result = self.__invoke(runner_args=args)
         return [
             f"exact:model.{str(x).split('.')[0]}.{str(x).split('.')[-1]}"
-            for x in r.result
+            for x in result
         ]
+
+    def get_artifacts_for_erd(self):
+        """Generate dbt artifacts using `dbt docs generate` command
+
+        Returns:
+            dbtRunnerResult: dbtRunnerResult
+        """
+        return self.__invoke(runner_args=["docs", "generate"])

dbterd/cli/params.py

@@ -97,6 +97,13 @@ def common_params(func):
         default=None,
         type=click.STRING,
     )
+    @click.option(
+        "--dbt-auto-artifacts",
+        help="Flag to force generating dbt artifact files leveraging Programmatic Invocation",
+        is_flag=True,
+        default=False,
+        show_default=True,
+    )
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
         return func(*args, **kwargs)  # pragma: no cover