Return ``(feature_view_names, feature_count)`` from resolved features. ``features`` is either a list of ``"feature_view:feature"`` strings or a ``FeatureService`` with ``feature_view_projections``. Returns: (fv_names, feat_count) where fv_names is a list of unique feature
(
features: Union[List[str], "feast.FeatureService"],
)
| 153 | |
| 154 | |
| 155 | def _parse_feature_info( |
| 156 | features: Union[List[str], "feast.FeatureService"], |
| 157 | ) -> tuple: |
| 158 | """Return ``(feature_view_names, feature_count)`` from resolved features. |
| 159 | |
| 160 | ``features`` is either a list of ``"feature_view:feature"`` strings or |
| 161 | a ``FeatureService`` with ``feature_view_projections``. |
| 162 | |
| 163 | Returns: |
| 164 | (fv_names, feat_count) where fv_names is a list of unique feature |
| 165 | view name strings and feat_count is the total number of features. |
| 166 | """ |
| 167 | from feast.feature_service import FeatureService |
| 168 | from feast.utils import _parse_feature_ref |
| 169 | |
| 170 | if isinstance(features, FeatureService): |
| 171 | projections = features.feature_view_projections |
| 172 | fv_names = [p.name for p in projections] |
| 173 | feat_count = sum(len(p.features) for p in projections) |
| 174 | elif isinstance(features, list): |
| 175 | feat_count = len(features) |
| 176 | fv_names = list({_parse_feature_ref(ref)[0] for ref in features if ":" in ref}) |
| 177 | else: |
| 178 | fv_names = [] |
| 179 | feat_count = 0 |
| 180 | return fv_names, feat_count |
| 181 | |
| 182 | |
| 183 | def _resolve_feature_counts( |