Method __init__

sdk/python/feast/feature_store.py:165–217

Creates a FeatureStore object. Args: repo_path (optional): Path to the feature repo. Defaults to the current working directory. config (optional): Configuration object used to configure the feature store. fs_yaml_file (optional): Path to the `fea

(
        self,
        repo_path: Optional[str] = None,
        config: Optional[RepoConfig] = None,
        fs_yaml_file: Optional[Path] = None,
    )

    _feature_service_cache: Dict[str, List[str]]

    def __init__(
        self,
        repo_path: Optional[str] = None,
        config: Optional[RepoConfig] = None,
        fs_yaml_file: Optional[Path] = None,
    ):
        """
        Creates a FeatureStore object.

        Args:
            repo_path (optional): Path to the feature repo. Defaults to the current working directory.
            config (optional): Configuration object used to configure the feature store.
            fs_yaml_file (optional): Path to the `feature_store.yaml` file used to configure the feature store.
                At most one of 'fs_yaml_file' and 'config' can be set.

        Raises:
            ValueError: If both or neither of repo_path and config are specified.
        """
        if fs_yaml_file is not None and config is not None:
            raise ValueError("You cannot specify both fs_yaml_file and config.")

        configure_ca_trust_store_env_variables()

        if repo_path:
            self.repo_path = Path(repo_path)
        else:
            self.repo_path = Path(os.getcwd())

        # If config is specified, or fs_yaml_file is specified, those take precedence over
        # the default feature_store.yaml location under repo_path.
        if config is not None:
            self.config = config
        elif fs_yaml_file is not None:
            self.config = load_repo_config(self.repo_path, fs_yaml_file)
        else:
            self.config = load_repo_config(
                self.repo_path, utils.get_default_yaml_file_path(self.repo_path)
            )

        # Initialize lazy-loaded components as None
        self._registry = None
        self._provider = None
        self._openlineage_emitter = None

        # Initialize feature service cache for performance optimization
        self._feature_service_cache = {}

        # Cache for _resolve_feature_service_name lookups
        self._fs_name_cache: Dict[frozenset, Optional[str]] = {}
        self._fs_name_index: Dict[frozenset, str] = {}
        self._fs_name_index_ts: float = -self._FS_NAME_INDEX_TTL_SECONDS

        self._mlflow_client: Any = _UNSET

    def _init_mlflow(self) -> Optional[Any]:
        """Bootstrap MLflow integration on first access.