(
self, request: RegistryServer_pb2.GetRegistryLineageRequest, context
)
| 1141 | return Empty() |
| 1142 | |
| 1143 | def GetRegistryLineage( |
| 1144 | self, request: RegistryServer_pb2.GetRegistryLineageRequest, context |
| 1145 | ): |
| 1146 | try: |
| 1147 | project = self.proxied_registry.get_project( |
| 1148 | name=request.project, allow_cache=True |
| 1149 | ) |
| 1150 | assert_permissions(resource=project, actions=[AuthzedAction.DESCRIBE]) |
| 1151 | except FeastObjectNotFoundException: |
| 1152 | pass |
| 1153 | direct_relationships, indirect_relationships = ( |
| 1154 | self.proxied_registry.get_registry_lineage( |
| 1155 | project=request.project, |
| 1156 | allow_cache=request.allow_cache, |
| 1157 | filter_object_type=request.filter_object_type, |
| 1158 | filter_object_name=request.filter_object_name, |
| 1159 | ) |
| 1160 | ) |
| 1161 | |
| 1162 | paginated_relationships, relationships_pagination = ( |
| 1163 | apply_pagination_and_sorting( |
| 1164 | direct_relationships, |
| 1165 | pagination=request.pagination, |
| 1166 | sorting=request.sorting, |
| 1167 | ) |
| 1168 | ) |
| 1169 | |
| 1170 | paginated_indirect_relationships, indirect_relationships_pagination = ( |
| 1171 | apply_pagination_and_sorting( |
| 1172 | indirect_relationships, |
| 1173 | pagination=request.pagination, |
| 1174 | sorting=request.sorting, |
| 1175 | ) |
| 1176 | ) |
| 1177 | |
| 1178 | return RegistryServer_pb2.GetRegistryLineageResponse( |
| 1179 | relationships=[rel.to_proto() for rel in paginated_relationships], |
| 1180 | indirect_relationships=[ |
| 1181 | rel.to_proto() for rel in paginated_indirect_relationships |
| 1182 | ], |
| 1183 | relationships_pagination=relationships_pagination, |
| 1184 | indirect_relationships_pagination=indirect_relationships_pagination, |
| 1185 | ) |
| 1186 | |
| 1187 | def GetObjectRelationships( |
| 1188 | self, request: RegistryServer_pb2.GetObjectRelationshipsRequest, context |
nothing calls this directly
no test coverage detected