Method do_get

sdk/python/feast/offline_server.py:233–267

(self, context: fl.ServerCallContext, ticket: fl.Ticket)

    @inject_user_details_decorator
    @arrow_server_error_handling_decorator
    def do_get(self, context: fl.ServerCallContext, ticket: fl.Ticket):
        key = ast.literal_eval(ticket.ticket.decode())
        if key not in self.flights:
            logger.error(f"Unknown key {key}")
            return None

        command = json.loads(key[1])

        self._validate_do_get_parameters(command)

        api = command["api"]
        logger.debug(f"get command is {command}")
        logger.debug(f"requested api is {api}")
        try:
            if api == OfflineServer.get_historical_features.__name__:
                table = self.get_historical_features(command, key).to_arrow()
            elif api == OfflineServer.pull_all_from_table_or_query.__name__:
                table = self.pull_all_from_table_or_query(command).to_arrow()
            elif api == OfflineServer.pull_latest_from_table_or_query.__name__:
                table = self.pull_latest_from_table_or_query(command).to_arrow()
            elif (
                api
                == OfflineServer.get_table_column_names_and_types_from_data_source.__name__
            ):
                table = self.get_table_column_names_and_types_from_data_source(command)
            else:
                raise NotImplementedError
        except Exception as e:
            logger.exception(e)
            traceback.print_exc()
            raise e

        # Get service is consumed, so we clear the corresponding flight and data
        del self.flights[key]
        return fl.RecordBatchStream(table)

    def _validate_offline_write_batch_parameters(self, command: dict):
        assert "feature_view_names" in command, (