(self, request, context)
| 62 | self._async_refresh() |
| 63 | |
| 64 | def Push(self, request, context): |
| 65 | try: |
| 66 | if request.features and request.typed_features: |
| 67 | context.set_code(grpc.StatusCode.INVALID_ARGUMENT) |
| 68 | context.set_details( |
| 69 | "Only one of features or typed_features may be set, not both" |
| 70 | ) |
| 71 | return PushResponse(status=False) |
| 72 | df = ( |
| 73 | parse_typed(request.typed_features) |
| 74 | if request.typed_features |
| 75 | else parse(request.features) |
| 76 | ) |
| 77 | if request.to == "offline": |
| 78 | to = PushMode.OFFLINE |
| 79 | elif request.to == "online": |
| 80 | to = PushMode.ONLINE |
| 81 | elif request.to == "online_and_offline": |
| 82 | to = PushMode.ONLINE_AND_OFFLINE |
| 83 | else: |
| 84 | raise ValueError( |
| 85 | f"{request.to} is not a supported push format. Please specify one of these ['online', 'offline', " |
| 86 | f"'online_and_offline']." |
| 87 | ) |
| 88 | self.fs.push( |
| 89 | push_source_name=request.stream_feature_view, |
| 90 | df=df, |
| 91 | allow_registry_cache=request.allow_registry_cache, |
| 92 | to=to, |
| 93 | ) |
| 94 | except PushSourceNotFoundException as e: |
| 95 | logger.exception(str(e)) |
| 96 | context.set_code(grpc.StatusCode.INVALID_ARGUMENT) |
| 97 | context.set_details(str(e)) |
| 98 | return PushResponse(status=False) |
| 99 | except Exception as e: |
| 100 | logger.exception(str(e)) |
| 101 | context.set_code(grpc.StatusCode.INTERNAL) |
| 102 | context.set_details(str(e)) |
| 103 | return PushResponse(status=False) |
| 104 | return PushResponse(status=True) |
| 105 | |
| 106 | def WriteToOnlineStore(self, request, context): |
| 107 | logger.warning( |
nothing calls this directly
no test coverage detected