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

Code before:
            raise ValueError(
                f"Manifest file not found in {os.path.join(self.project_dir, self.target_path, 'manifest.json')}"
            )

    def _get_node_prefect_config(
        self, manifest_node: Union[ManifestNode, SourceDefinition]
    ) -> dict[str, dict[str, Any]]:
        if isinstance(manifest_node, SourceDefinition):
            return manifest_node.meta.get("prefect", {})

        return manifest_node.config.meta.get("prefect", {})

    def _get_upstream_manifest_nodes_and_configs(
        self,
        manifest_node: ManifestNode,
    ) -> list[tuple[Union[ManifestNode, SourceDefinition], dict[str, Any]]]:
        """
        Get upstream nodes for a given node.
        Ephemeral nodes are traversed recursively to find non-ephemeral dependencies.
        """
        upstream_manifest_nodes: list[
            tuple[Union[ManifestNode, SourceDefinition], dict[str, Any]]
        ] = []
        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),

Review comment (suggestion):
`depends_manifest_node` might be a `ManifestNode` or a `SourceDefinition` right?

```suggestion
        def collect(node: ManifestNode | SourceDefinition):
```

Code after:
            raise ValueError(
                f"Manifest file not found in {os.path.join(self.project_dir, self.target_path, 'manifest.json')}"
            )

    def _get_node_prefect_config(
        self, manifest_node: Union[ManifestNode, SourceDefinition]
    ) -> dict[str, dict[str, Any]]:
        if isinstance(manifest_node, SourceDefinition):
            return manifest_node.meta.get("prefect", {})

        return manifest_node.config.meta.get("prefect", {})

    def _get_upstream_manifest_nodes_and_configs(
        self,
        manifest_node: ManifestNode,
    ) -> list[tuple[Union[ManifestNode, SourceDefinition], dict[str, Any]]]:
        """
        Get upstream nodes for a given node.
        Ephemeral nodes are traversed recursively to find non-ephemeral dependencies.
        Sources without relation_name are ignored.
        """
        upstream_manifest_nodes: list[
            tuple[Union[ManifestNode, SourceDefinition], dict[str, Any]]
        ] = []
        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"