* [Remote](reference/online-stores/remote.md)

{% content-ref url="remote.md" %}

[remote.md](remote.md)

{% endcontent-ref %}

This remote online store will let you interact with remote feature server. At this moment this only supports the read operation. You can use this online store and able retrieve online features `store.get_online_features` from remote feature server.

The registry is pointing to registry of remote feature store. If it is not accessible then should be configured to use remote registry.

{% code title="feature_store.yaml" %}

project: my-local-project

registry: /remote/data/registry.db

provider: local

online_store:

path: http://localhost:6566

type: remote

entity_key_serialization_version: 2

uv = ">=0.1.39,<0.2"

[feature.py311.dependencies]

python = "~=3.11.0"

py310 = ["py310"]

py311 = ["py311"]

import json

import logging

from datetime import datetime

from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Tuple

import requests

from pydantic import StrictStr

from feast import Entity, FeatureView, RepoConfig

from feast.infra.online_stores.online_store import OnlineStore

from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto

from feast.protos.feast.types.Value_pb2 import Value as ValueProto

from feast.repo_config import FeastConfigBaseModel

from feast.type_map import python_values_to_proto_values

from feast.value_type import ValueType

logger = logging.getLogger(__name__)

class RemoteOnlineStoreConfig(FeastConfigBaseModel):

"""Remote Online store config for remote online store"""

type: Literal["remote"] = "remote"

"""Online store type selector"""

path: StrictStr = "http://localhost:6566"

""" str: Path to metadata store.

class RemoteOnlineStore(OnlineStore):

"""

def online_write_batch(

self,

config: RepoConfig,

table: FeatureView,

data: List[

Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]

],

progress: Optional[Callable[[int], Any]],

) -> None:

pass

def online_read(

self,

config: RepoConfig,

table: FeatureView,

entity_keys: List[EntityKeyProto],

requested_features: Optional[List[str]] = None,

) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:

assert isinstance(config.online_store, RemoteOnlineStoreConfig)

config.online_store.__class__ = RemoteOnlineStoreConfig

req_body = self._construct_online_read_api_json_request(

entity_keys, table, requested_features

)

response = requests.post(

f"{config.online_store.path}/get-online-features", data=req_body

)

if response.status_code == 200:

logger.debug("Able to retrieve the online features from feature server.")

response_json = json.loads(response.text)

event_ts = self._get_event_ts(response_json)

# Iterating over results and converting the API results in column format to row format.

result_tuples: List[

Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]

] = []

for feature_value_index in range(len(entity_keys)):

feature_values_dict: Dict[str, ValueProto] = dict()

for index, feature_name in enumerate(

response_json["metadata"]["feature_names"]

):

if (

requested_features is not None

and feature_name in requested_features

):

if (

response_json["results"][index]["statuses"][

feature_value_index

]

== "PRESENT"

):

message = python_values_to_proto_values(

[

response_json["results"][index]["values"][

feature_value_index

]

],

ValueType.UNKNOWN,

)

feature_values_dict[feature_name] = message[0]

else:

feature_values_dict[feature_name] = ValueProto()

result_tuples.append((event_ts, feature_values_dict))

return result_tuples

else:

error_msg = f"Unable to retrieve the online store data using feature server API. Error_code={response.status_code}, error_message={response.reason}"

logger.error(error_msg)

raise RuntimeError(error_msg)

def _construct_online_read_api_json_request(

self,

entity_keys: List[EntityKeyProto],

table: FeatureView,

requested_features: Optional[List[str]] = None,

):

api_requested_features = []

if requested_features is not None:

for requested_feature in requested_features:

api_requested_features.append(f"{table.name}:{requested_feature}")

entity_values = []

entity_key = ""

for row in entity_keys:

entity_key = row.join_keys[0]

entity_values.append(

getattr(row.entity_values[0], row.entity_values[0].WhichOneof("val"))

)

req_body = json.dumps(

{

"features": api_requested_features,

"entities": {entity_key: entity_values},

}

)

return req_body

def _check_if_feature_requested(self, feature_name, requested_features):

for requested_feature in requested_features:

if feature_name in requested_feature:

return True

return False

def _get_event_ts(self, response_json) -> datetime:

event_ts = ""

if len(response_json["results"]) > 1:

event_ts = response_json["results"][1]["event_timestamps"][0]

return datetime.fromisoformat(event_ts.replace("Z", "+00:00"))

def update(

self,

config: RepoConfig,

tables_to_delete: Sequence[FeatureView],

tables_to_keep: Sequence[FeatureView],

entities_to_delete: Sequence[Entity],

entities_to_keep: Sequence[Entity],

partial: bool,

):

pass

def teardown(

self,

config: RepoConfig,

tables: Sequence[FeatureView],

entities: Sequence[Entity],

):

pass

"remote": "feast.infra.online_stores.remote.RemoteOnlineStore",

from tests.integration.feature_repos.integration_test_repo_config import (

IntegrationTestRepoConfig, # noqa: E402

from tests.integration.feature_repos.universal.data_sources.file import (

FileDataSourceCreator, # noqa: E402

fs.serve(

host="localhost",

port=port,

no_access_log=True,

keep_alive_timeout=30,

type_="http",

workers=1,

registry_ttl_sec=30,

)

] = {"sqlite": ({"type": "sqlite"}, None)}