Method parse

sdk/python/feast/dbt/parser.py:82–116

Load and parse the manifest.json file using dbt-artifacts-parser. Raises: FileNotFoundError: If manifest.json doesn't exist ValueError: If manifest.json is invalid JSON ImportError: If dbt-artifacts-parser is not installed

(self)

        self._parsed_manifest: Optional[Any] = None

    def parse(self) -> None:
        """
        Load and parse the manifest.json file using dbt-artifacts-parser.

        Raises:
            FileNotFoundError: If manifest.json doesn't exist
            ValueError: If manifest.json is invalid JSON
            ImportError: If dbt-artifacts-parser is not installed
        """
        if not self.manifest_path.exists():
            raise FileNotFoundError(
                f"dbt manifest not found at {self.manifest_path}.\n"
                f"Run 'dbt compile' or 'dbt run' first.\n"
                f"Expected path: <dbt_project>/target/manifest.json"
            )

        try:
            with open(self.manifest_path, "r") as f:
                self._raw_manifest = json.load(f)
        except json.JSONDecodeError as e:
            raise ValueError(
                f"Invalid JSON in manifest: {e}\nTry: dbt clean && dbt compile"
            )

        # Parse using dbt-artifacts-parser for typed access
        try:
            from dbt_artifacts_parser.parser import parse_manifest

            assert self._raw_manifest is not None
            self._parsed_manifest = parse_manifest(manifest=self._raw_manifest)
        except ImportError:
            raise ImportError(
                "dbt-artifacts-parser is required for dbt integration.\n"
                "Install with: pip install 'feast[dbt]' or pip install dbt-artifacts-parser"
            )

    def _extract_column_from_node(self, col_name: str, col_data: Any) -> DbtColumn:
        """Extract column info from a parsed node column."""