Skip to content

base

Base classes for feature store integrations.

BaseFeatureStore

Bases: ABC

Abstract base class for feature store integrations.

Provides a unified interface for saving and retrieving engineered features from various feature stores.

Parameters:

Name Type Description Default
config FeatureStoreConfig

Configuration for the feature store

 required

Examples:

>>> store = ConcreteFeatureStore(config)
>>> store.save_features(X_transformed, feature_set, feature_view_name='my_features')
>>> features = store.get_features(entity_df, feature_names=['feat1', 'feat2'])
Source code in featcopilot/stores/base.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class BaseFeatureStore(ABC):
    """
    Abstract base class for feature store integrations.

    Provides a unified interface for saving and retrieving
    engineered features from various feature stores.

    Parameters
    ----------
    config : FeatureStoreConfig
        Configuration for the feature store

    Examples
    --------
    >>> store = ConcreteFeatureStore(config)
    >>> store.save_features(X_transformed, feature_set, feature_view_name='my_features')
    >>> features = store.get_features(entity_df, feature_names=['feat1', 'feat2'])
    """

    def __init__(self, config: FeatureStoreConfig):
        self.config = config
        self._is_initialized = False

    @abstractmethod
    def initialize(self) -> None:
        """
        Initialize connection to the feature store.

        This should be called before any other operations.
        """
        pass

    @abstractmethod
    def save_features(
        self,
        df: pd.DataFrame,
        feature_set: Optional[FeatureSet] = None,
        feature_view_name: str = "featcopilot_features",
        description: Optional[str] = None,
        **kwargs,
    ) -> None:
        """
        Save features to the feature store.

        Parameters
        ----------
        df : DataFrame
            DataFrame containing features to save
        feature_set : FeatureSet, optional
            FeatCopilot FeatureSet with metadata
        feature_view_name : str
            Name for the feature view/table
        description : str, optional
            Description of the feature view
        **kwargs
            Additional store-specific options
        """
        pass

    @abstractmethod
    def get_features(
        self,
        entity_df: pd.DataFrame,
        feature_names: list[str],
        feature_view_name: str = "featcopilot_features",
        **kwargs,
    ) -> pd.DataFrame:
        """
        Retrieve features from the feature store.

        Parameters
        ----------
        entity_df : DataFrame
            DataFrame with entity keys and timestamps
        feature_names : list
            Names of features to retrieve
        feature_view_name : str
            Name of the feature view/table
        **kwargs
            Additional store-specific options

        Returns
        -------
        DataFrame
            DataFrame with requested features
        """
        pass

    @abstractmethod
    def list_feature_views(self) -> list[str]:
        """
        List all feature views in the store.

        Returns
        -------
        list
            Names of feature views
        """
        pass

    @abstractmethod
    def get_feature_view_schema(self, feature_view_name: str) -> dict[str, Any]:
        """
        Get schema/metadata for a feature view.

        Parameters
        ----------
        feature_view_name : str
            Name of the feature view

        Returns
        -------
        dict
            Schema information
        """
        pass

    @abstractmethod
    def delete_feature_view(self, feature_view_name: str) -> bool:
        """
        Delete a feature view.

        Parameters
        ----------
        feature_view_name : str
            Name of the feature view to delete

        Returns
        -------
        bool
            Whether deletion was successful
        """
        pass

    def close(self) -> None:
        """Close connection to the feature store."""
        self._is_initialized = False

    def __enter__(self):
        self.initialize()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        return False

close()

Close connection to the feature store.

Source code in featcopilot/stores/base.py 
156
157
158
def close(self) -> None:
    """Close connection to the feature store."""
    self._is_initialized = False

delete_feature_view(feature_view_name) abstractmethod

Delete a feature view.

Parameters:

Name Type Description Default
feature_view_name str

Name of the feature view to delete

 required

Returns:

Type Description
bool

Whether deletion was successful
Source code in featcopilot/stores/base.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@abstractmethod
def delete_feature_view(self, feature_view_name: str) -> bool:
    """
    Delete a feature view.

    Parameters
    ----------
    feature_view_name : str
        Name of the feature view to delete

    Returns
    -------
    bool
        Whether deletion was successful
    """
    pass

get_feature_view_schema(feature_view_name) abstractmethod

Get schema/metadata for a feature view.

Parameters:

Name Type Description Default
feature_view_name str

Name of the feature view

 required

Returns:

Type Description
dict

Schema information
Source code in featcopilot/stores/base.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@abstractmethod
def get_feature_view_schema(self, feature_view_name: str) -> dict[str, Any]:
    """
    Get schema/metadata for a feature view.

    Parameters
    ----------
    feature_view_name : str
        Name of the feature view

    Returns
    -------
    dict
        Schema information
    """
    pass

get_features(entity_df, feature_names, feature_view_name='featcopilot_features', **kwargs) abstractmethod

Retrieve features from the feature store.

Parameters:

Name Type Description Default
entity_df DataFrame

DataFrame with entity keys and timestamps

 required
feature_names list

Names of features to retrieve

 required
feature_view_name str

Name of the feature view/table

 'featcopilot_features'
**kwargs

Additional store-specific options

 {}

Returns:

Type Description
DataFrame

DataFrame with requested features
Source code in featcopilot/stores/base.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@abstractmethod
def get_features(
    self,
    entity_df: pd.DataFrame,
    feature_names: list[str],
    feature_view_name: str = "featcopilot_features",
    **kwargs,
) -> pd.DataFrame:
    """
    Retrieve features from the feature store.

    Parameters
    ----------
    entity_df : DataFrame
        DataFrame with entity keys and timestamps
    feature_names : list
        Names of features to retrieve
    feature_view_name : str
        Name of the feature view/table
    **kwargs
        Additional store-specific options

    Returns
    -------
    DataFrame
        DataFrame with requested features
    """
    pass

initialize() abstractmethod

Initialize connection to the feature store.

This should be called before any other operations.

Source code in featcopilot/stores/base.py 
45
46
47
48
49
50
51
52
@abstractmethod
def initialize(self) -> None:
    """
    Initialize connection to the feature store.

    This should be called before any other operations.
    """
    pass

list_feature_views() abstractmethod

List all feature views in the store.

Returns:

Type Description
list

Names of feature views
Source code in featcopilot/stores/base.py 
110
111
112
113
114
115
116
117
118
119
120
@abstractmethod
def list_feature_views(self) -> list[str]:
    """
    List all feature views in the store.

    Returns
    -------
    list
        Names of feature views
    """
    pass

save_features(df, feature_set=None, feature_view_name='featcopilot_features', description=None, **kwargs) abstractmethod

Save features to the feature store.

Parameters:

Name Type Description Default
df DataFrame

DataFrame containing features to save

 required
feature_set FeatureSet

FeatCopilot FeatureSet with metadata

 None
feature_view_name str

Name for the feature view/table

 'featcopilot_features'
description str

Description of the feature view

 None
**kwargs

Additional store-specific options

 {}
Source code in featcopilot/stores/base.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@abstractmethod
def save_features(
    self,
    df: pd.DataFrame,
    feature_set: Optional[FeatureSet] = None,
    feature_view_name: str = "featcopilot_features",
    description: Optional[str] = None,
    **kwargs,
) -> None:
    """
    Save features to the feature store.

    Parameters
    ----------
    df : DataFrame
        DataFrame containing features to save
    feature_set : FeatureSet, optional
        FeatCopilot FeatureSet with metadata
    feature_view_name : str
        Name for the feature view/table
    description : str, optional
        Description of the feature view
    **kwargs
        Additional store-specific options
    """
    pass

FeatureStoreConfig

Bases: BaseModel

Base configuration for feature stores.

Source code in featcopilot/stores/base.py 
12
13
14
15
16
17
18
19
class FeatureStoreConfig(BaseModel):
    """Base configuration for feature stores."""

    name: str = Field(description="Feature store name")
    entity_columns: list[str] = Field(default_factory=list, description="Entity/key columns")
    timestamp_column: Optional[str] = Field(default=None, description="Event timestamp column")
    feature_prefix: str = Field(default="", description="Prefix for feature names")
    tags: dict[str, str] = Field(default_factory=dict, description="Tags/labels for features")