Repository: PrefectHQ/prefect
File: src/integrations/prefect-dbt/prefect_dbt/core/runner.py (Python)

Code before:
        visited: set[str] = set()

        def collect(node: ManifestNode):
            for depends_on_node in node.depends_on_nodes:  # type: ignore[reportUnknownMemberType]
                if depends_on_node in visited:
                    continue
                visited.add(depends_on_node)

                depends_manifest_node = (
                    self.manifest.nodes.get(depends_on_node)  # type: ignore[reportUnknownMemberType]
                    or self.manifest.sources.get(depends_on_node)  # type: ignore[reportUnknownMemberType]
                )

                if not depends_manifest_node:
                    continue

                # recursive for ephemerals
                if (
                    not depends_manifest_node.relation_name
                    and depends_manifest_node.is_ephemeral
                ):
                    collect(depends_manifest_node)
                    continue

                upstream_manifest_nodes.append(
                    (
                        depends_manifest_node,
                        self._get_node_prefect_config(depends_manifest_node),
                    )
                )

        collect(manifest_node)
        return upstream_manifest_nodes

    def _get_compiled_code_path(self, manifest_node: ManifestNode) -> Path:
        """Get the path to compiled code for a manifest node."""
        return (
            Path(self.project_dir)
            / self.target_path
            / "compiled"
            / self.project_name
            / manifest_node.original_file_path
        )

    def _get_compiled_code(
        self, manifest_node: Union[ManifestNode, SourceDefinition]
    ) -> str:
        """Get compiled code for a manifest node if it exists and is enabled."""
        if not self.include_compiled_code or isinstance(
            manifest_node, SourceDefinition
        ):

Review comment (suggestion):
is it worth adding a guard here?
```suggestion
                if (
                isinstance(depends_manifest_node, ManifestNode)
```

Code after:
        ] = []
        visited: set[str] = set()

        def collect(node: ManifestNode | SourceDefinition):
            for depends_on_node in node.depends_on_nodes:  # type: ignore[reportUnknownMemberType]
                if depends_on_node in visited:
                    continue
                visited.add(depends_on_node)

                depends_manifest_node = (
                    self.manifest.nodes.get(depends_on_node)  # type: ignore[reportUnknownMemberType]
                    or self.manifest.sources.get(depends_on_node)  # type: ignore[reportUnknownMemberType]
                )

                if not depends_manifest_node:
                    continue

                # sources without relation_name
                if (
                    depends_on_node in self.manifest.sources
                    and not depends_manifest_node.relation_name
                ):
                    continue

                # recursive for ephemerals
                if (
                    depends_on_node in self.manifest.nodes
                    and depends_manifest_node.config.materialized == "ephemeral"
                ):
                    collect(depends_manifest_node)
                    continue

                upstream_manifest_nodes.append(
                    (
                        depends_manifest_node,
                        self._get_node_prefect_config(depends_manifest_node),
                    )
                )

        collect(manifest_node)
        return upstream_manifest_nodes

    def _get_compiled_code_path(self, manifest_node: ManifestNode) -> Path:
        """Get the path to compiled code for a manifest node."""
        return (
            Path(self.project_dir)
            / self.target_path
            / "compiled"
            / self.project_name
            / manifest_node.original_file_path
        )