(
feast_type: list[FeastObject],
p: Permission,
policy_node: Node,
store: FeatureStore,
tags_filter: Optional[dict[str, str]],
)
| 154 | |
| 155 | |
| 156 | def handle_fv_verbose_permissions_command( |
| 157 | feast_type: list[FeastObject], |
| 158 | p: Permission, |
| 159 | policy_node: Node, |
| 160 | store: FeatureStore, |
| 161 | tags_filter: Optional[dict[str, str]], |
| 162 | ): |
| 163 | feature_views = [] |
| 164 | feature_views_names = set() |
| 165 | if feast_type == FeatureView: |
| 166 | feature_views = store.list_all_feature_views(tags=tags_filter) # type: ignore[assignment] |
| 167 | elif feast_type == OnDemandFeatureView: |
| 168 | feature_views = store.list_on_demand_feature_views( |
| 169 | tags=tags_filter # type: ignore[assignment] |
| 170 | ) |
| 171 | elif feast_type == BatchFeatureView: |
| 172 | feature_views = store.list_batch_feature_views(tags=tags_filter) # type: ignore[assignment] |
| 173 | elif feast_type == StreamFeatureView: |
| 174 | feature_views = store.list_stream_feature_views( |
| 175 | tags=tags_filter # type: ignore[assignment] |
| 176 | ) |
| 177 | for fv in feature_views: |
| 178 | if p.match_resource(fv): # type: ignore[arg-type] |
| 179 | feature_views_names.add(fv.name) |
| 180 | if len(feature_views_names) > 0: |
| 181 | Node( |
| 182 | feast_type.__name__ + " " + str(list(feature_views_names)), # type: ignore[union-attr, attr-defined] |
| 183 | parent=policy_node, |
| 184 | ) |
| 185 | else: |
| 186 | Node(feast_type.__name__ + ": none", parent=policy_node) # type: ignore[union-attr, attr-defined] |
| 187 | |
| 188 | |
| 189 | def handle_not_verbose_permissions_command( |
nothing calls this directly
no test coverage detected