| SQLite | [x] | [ ] |

Note: SQLite is in limited access and only working on Python 3.10. It will be updated as [sqlite_vec](https://github.com/asg017/sqlite-vec/) progresses.

We offer two Online Store options for Vector Databases. PGVector and SQLite.

If you are using `pyenv` to manage your Python versions, you can install the SQLite extension with the following command:

PYTHON_CONFIGURE_OPTS="--enable-loadable-sqlite-extensions" \

LDFLAGS="-L/opt/homebrew/opt/sqlite/lib" \

CPPFLAGS="-I/opt/homebrew/opt/sqlite/include" \

pyenv install 3.10.14

And you can the Feast install package via:

pip install feast[sqlite_vec]

{% 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

platforms = ["linux-64", "osx-arm64"]

uv = ">=0.1.39,<0.2"

[feature.py311.dependencies]

python = "~=3.11.0"

py310 = ["py310"]

py311 = ["py311"]

import logging

def list_all_feature_views(

self, allow_cache: bool = False

) -> List[Union[FeatureView, StreamFeatureView, OnDemandFeatureView]]:

"""

return self._list_all_feature_views(allow_cache)

logging.warning(

"list_feature_views will make breaking changes. Please use list_batch_feature_views instead. "

"list_feature_views will behave like list_all_feature_views in the future."

)

def _list_all_feature_views(

self,

allow_cache: bool = False,

) -> List[Union[FeatureView, StreamFeatureView, OnDemandFeatureView]]:

all_feature_views = (

self._list_feature_views(allow_cache)

+ self._list_stream_feature_views(allow_cache)

+ self.list_on_demand_feature_views(allow_cache)

)

return all_feature_views

) -> List[FeatureView]:

logging.warning(

"_list_feature_views will make breaking changes. Please use _list_batch_feature_views instead. "

"_list_feature_views will behave like _list_all_feature_views in the future."

)

feature_views = []

for fv in self._registry.list_feature_views(

self.project, allow_cache=allow_cache

):

if (

hide_dummy_entity

and fv.entities

and fv.entities[0] == DUMMY_ENTITY_NAME

):

fv.entities = []

fv.entity_columns = []

feature_views.append(fv)

return feature_views

def _list_batch_feature_views(

self,

allow_cache: bool = False,

hide_dummy_entity: bool = True,

available_feature_views,

requested_feature_view_name = (

feature.split(":")[0] if isinstance(feature, str) else feature

)

for feature_view in available_feature_views:

if feature_view.name == requested_feature_view_name:

requested_feature_view = feature_view

if not requested_feature_view:

raise ValueError(

f"Feature view {requested_feature_view} not found in the registry."

)

requested_feature_view,

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