feat: dlt.Schema.to_dot() graphviz export

dlt/cli/command_wrappers.py

@@ -145,6 +145,8 @@ def schema_command_wrapper(file_path: str, format_: str, remove_defaults: bool)
         schema_str = s.to_pretty_yaml(remove_defaults=remove_defaults)
     elif format_ == "dbml":
         schema_str = s.to_dbml()
+    elif format_ == "dot":
+        schema_str = s.to_dot()
     else:
         schema_str = s.to_pretty_yaml(remove_defaults=remove_defaults)
 

dlt/cli/pipeline_command.py

@@ -354,6 +354,8 @@ def _display_pending_packages() -> Tuple[Sequence[str], Sequence[str]]:
             schema_str = s.to_pretty_yaml(remove_defaults=remove_defaults_)
         elif format_ == "dbml":
             schema_str = s.to_dbml()
+        elif format_ == "dot":
+            schema_str = s.to_dot()
         else:
             schema_str = s.to_pretty_yaml(remove_defaults=remove_defaults_)
 

dlt/common/schema/schema.py

@@ -688,6 +688,18 @@ def __repr__(self) -> str:
         }
         return simple_repr("dlt.Schema", **without_none(kwargs))
 
+    def _repr_html_(self, **kwargs: Any) -> str:
+        """Render the Schema has a graphviz graph and display it using HTML
+
+        This method is automatically called by notebooks renderers (IPython, marimo, etc.)
+        ref: https://ipython.readthedocs.io/en/stable/config/integrating.html
+
+        `dlt.helpers.graphviz.render_with_html()` has not external Python or system dependencies.
+        """
+        from dlt.helpers.graphviz import _render_dot_with_html
+
+        return _render_dot_with_html(self.to_dot(**kwargs))
+
     def to_dict(
         self,
         remove_defaults: bool = False,
@@ -768,6 +780,49 @@ def to_dbml(
         )
         return str(dbml_schema.dbml)
 
+    def to_dot(
+        self,
+        remove_processing_hints: bool = False,
+        include_dlt_tables: bool = True,
+        include_internal_dlt_ref: bool = True,
+        include_parent_child_ref: bool = True,
+        include_root_child_ref: bool = True,
+        group_by_resource: bool = False,
+    ) -> str:
+        """Convert schema to a Graphviz DOT string.
+
+        Args:
+            remove_processing_hints: If True, remove hints used for data processing and redundant information.
+                This reduces the size of the schema and improves readability.
+            include_dlt_tables: If True, include data tables and internal dlt tables. This will influence table
+                references and groups produced.
+            include_internal_dlt_ref: If True, include references between tables `_dlt_version`, `_dlt_loads` and `_dlt_pipeline_state`
+            include_parent_child_ref: If True, include references from `child._dlt_parent_id` to `parent._dlt_id`
+            include_root_child_ref: If True, include references from `child._dlt_root_id` to `root._dlt_id`
+            group_by_resource: If True, group tables by resource and create subclusters.
+
+        Returns:
+            A DOT string of the schema
+        """
+        from dlt.helpers.graphviz import schema_to_graphviz
+
+        stored_schema = self.to_dict(
+            # setting this to `True` removes `name` fields that are used in `schema_to_dbml()`
+            # if required, we can refactor `dlt.helpers.dbml` to support this
+            remove_defaults=False,
+            remove_processing_hints=remove_processing_hints,
+        )
+
+        dot = schema_to_graphviz(
+            stored_schema,
+            include_dlt_tables=include_dlt_tables,
+            include_internal_dlt_ref=include_internal_dlt_ref,
+            include_parent_child_ref=include_parent_child_ref,
+            include_root_child_ref=include_root_child_ref,
+            group_by_resource=group_by_resource,
+        )
+        return dot
+
     def clone(
         self,
         with_name: str = None,

dlt/common/storages/configuration.py

@@ -22,7 +22,7 @@
 from dlt.common.utils import digest128
 
 
-TSchemaFileFormat = Literal["json", "yaml", "dbml"]
+TSchemaFileFormat = Literal["json", "yaml", "dbml", "dot"]
 SCHEMA_FILES_EXTENSIONS = get_args(TSchemaFileFormat)
 
 

dlt/common/storages/schema_storage.py

@@ -281,7 +281,9 @@ def _parse_schema_str(schema_str: str, extension: TSchemaFileFormat) -> DictStrA
         elif extension == "yaml":
             imported_schema = yaml.safe_load(schema_str)
         elif extension == "dbml":
-            raise ValueError(extension, "Schema parser for dbml not yet implemented")
+            raise ValueError(extension, "Schema parser for `dbml` not yet implemented")
+        elif extension == "dot":
+            raise ValueError(extension, "Schema parser for `dot` not yet implemented")
         else:
             raise ValueError(extension)
         return imported_schema