def create_fg_from_df()

in utilities/feature_store_helper.py [0:0]

• 32 lines of code
• 8 McCabe index (conditional complexity)

    def create_fg_from_df(self, fg_name: str, df: pd.DataFrame, description: str=None,
                          id_name: str='Id', event_time_name: str='UpdateTime', 
                          tags: Dict=None, online: bool=True, s3_uri: str=None) -> None:
        """Creates a new feature group based on a Pandas dataframe.
        
        This method creates a new feature group using feature definitions derived from 
        the column names and column data types of an existing Pandas dataframe. This is
        a convenience method to avoid having to know the details of FeatureDefinition 
        syntax and data types.
        
        Args:
            fg_name (str): Name of the feature group to create.
            df (pd.DataFrame): Pandas dataframe.
            description (str): Text description that should be associated with this new feature group.
            id_name (str): Name of feature that will be used as the record identifier. Must exist in the dataframe.
            even_time_name (str): Name of feature that will be used as the event time. Must exist in the dataframe.
            tags (dict): Dictionary of tags to associate with the new feature group.
            online (bool): Whether or not this feature group will have an online store.
            s3_uri (str): Base s3 uri indicating where the offline storage should be kept for this feature group.
        """

        if not id_name in df.columns:
            print(f'invalid id column name: {id_name}')
            return
        if not event_time_name in df.columns:
            print(f'invalid event time column name: {event_time_name}')
            return

        if s3_uri is None:
            s3_uri = f's3://{self._default_bucket}/offline-store'

        other_args = {}
        if s3_uri is not None:
            other_args['OfflineStoreConfig'] = {'S3StorageConfig': {'S3Uri': s3_uri}}

        if tags is not None:
            tags_as_kv_array = []
            for k, v in tags.items():
                curr_kv = {'Key': k, 'Value': self._escape_tag_chars(v)}
                tags_as_kv_array.append(curr_kv)
            other_args['Tags'] = tags_as_kv_array
        
        if description is not None:
            other_args['Description'] = description

        resp = self._sm_client.create_feature_group(
                FeatureGroupName = fg_name,
                RecordIdentifierFeatureName = id_name,
                EventTimeFeatureName = event_time_name,
                FeatureDefinitions = self._df_to_feature_defs(df),
                OnlineStoreConfig = {'EnableOnlineStore': online},
                RoleArn = self._role,
                **other_args)

        self._wait_for_feature_group_creation_complete(fg_name)
        return