fix(ingestion/looker): skip personal folder independent looks

metadata-ingestion/src/datahub/ingestion/source/looker/looker_source.py

@@ -286,7 +286,8 @@ def _get_views_from_fields(self, fields: List[str]) -> List[str]:
             except AssertionError:
                 self.reporter.report_warning(
                     title="Failed to Extract View Name from Field",
-                    message="The field was not prefixed by a view name. This can happen when the field references another dynamic field.",
+                    message="The field was not prefixed by a view name. This can happen when the field references "
+                    "another dynamic field.",
                     context=f"Field Name: {field_name}",
                 )
                 continue
@@ -380,15 +381,16 @@ def _get_input_fields_from_query(
                 )
             )
 
-        # A query uses fields for filtering and those fields are defined in views, find the views those fields use
+        # A query uses fields for filtering, and those fields are defined in views, find the views those fields use
         filters: MutableMapping[str, Any] = (
             query.filters if query.filters is not None else {}
         )
         for field in filters.keys():
             if field is None:
                 continue
 
-            # we haven't loaded in metadata about the explore yet, so we need to wait until explores are populated later to fetch this
+            # we haven't loaded in metadata about the explore yet, so we need to wait until explores are populated
+            # later to fetch this
             result.append(
                 InputFieldElement(
                     name=field, view_field=None, model=query.model, explore=query.view
@@ -1486,13 +1488,27 @@ def extract_independent_looks(self) -> Iterable[MetadataWorkUnit]:
         )
         for look in all_looks:
             if look.id in self.reachable_look_registry:
-                # This look is reachable from Dashboard
+                # This look is reachable from the Dashboard
                 continue
 
             if look.query_id is None:
                 logger.info(f"query_id is None for look {look.title}({look.id})")
                 continue
 
+            if self.source_config.skip_personal_folders:
+                if look.folder is not None and (
+                    look.folder.is_personal or look.folder.is_personal_descendant
+                ):
+                    self.reporter.info(
+                        title="Dropped Look",
+                        message="Dropped due to being a personal folder",
+                        context=f"Look ID: {look.id}",
+                    )
+
+                    assert look.id, "Looker id is null"
+                    self.reporter.report_charts_dropped(look.id)
+                    continue
+
             if look.id is not None:
                 query: Optional[Query] = self.looker_api.get_look(
                     look.id, fields=["query"]
@@ -1510,11 +1526,12 @@ def extract_independent_looks(self) -> Iterable[MetadataWorkUnit]:
                 LookerDashboardElement
             ] = self._get_looker_dashboard_element(
                 DashboardElement(
-                    id=f"looks_{look.id}",  # to avoid conflict with non-standalone looks (element.id prefixes), we add the "looks_" prefix to look.id.
+                    id=f"looks_{look.id}",  # to avoid conflict with non-standalone looks (element.id prefixes),
+                    # we add the "looks_" prefix to look.id.
                     title=look.title,
                     subtitle_text=look.description,
                     look_id=look.id,
-                    dashboard_id=None,  # As this is independent look
+                    dashboard_id=None,  # As this is an independent look
                     look=LookWithQuery(
                         query=query, folder=look.folder, user_id=look.user_id
                     ),