import logging

import os

import uuid

from binascii import hexlify

from datetime import datetime, timedelta, timezone

from enum import Enum

from threading import Lock

from typing import Any, Callable, List, Literal, Optional, Union, cast

from pydantic import ConfigDict, Field, StrictStr

import feast

from feast.base_feature_view import BaseFeatureView

from feast.data_source import DataSource

from feast.entity import Entity

from feast.errors import (

DataSourceObjectNotFoundException,

EntityNotFoundException,

FeatureServiceNotFoundException,

FeatureViewNotFoundException,

PermissionNotFoundException,

ProjectNotFoundException,

ProjectObjectNotFoundException,

SavedDatasetNotFound,

ValidationReferenceNotFound,

)

from feast.feature_service import FeatureService

from feast.feature_view import FeatureView

from feast.infra.infra_object import Infra

from feast.infra.registry import proto_registry_utils

from feast.infra.registry.base_registry import BaseRegistry

from feast.infra.utils.snowflake.snowflake_utils import (

GetSnowflakeConnection,

execute_snowflake_statement,

)

from feast.labeling.label_view import LabelView

from feast.on_demand_feature_view import OnDemandFeatureView

from feast.permissions.permission import Permission

from feast.project import Project

from feast.project_metadata import ProjectMetadata

from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto

from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto

from feast.protos.feast.core.FeatureService_pb2 import (

FeatureService as FeatureServiceProto,

)

from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto

from feast.protos.feast.core.InfraObject_pb2 import Infra as InfraProto

from feast.protos.feast.core.LabelView_pb2 import LabelView as LabelViewProto

from feast.protos.feast.core.OnDemandFeatureView_pb2 import (

OnDemandFeatureView as OnDemandFeatureViewProto,

)

from feast.protos.feast.core.Permission_pb2 import Permission as PermissionProto

from feast.protos.feast.core.Project_pb2 import Project as ProjectProto

from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto

from feast.protos.feast.core.SavedDataset_pb2 import SavedDataset as SavedDatasetProto

from feast.protos.feast.core.StreamFeatureView_pb2 import (

StreamFeatureView as StreamFeatureViewProto,

)

from feast.protos.feast.core.ValidationProfile_pb2 import (

ValidationReference as ValidationReferenceProto,

)

from feast.repo_config import RegistryConfig

from feast.saved_dataset import SavedDataset, ValidationReference

from feast.stream_feature_view import StreamFeatureView

from feast.utils import _utc_now, has_all_tags, to_naive_utc

logger = logging.getLogger(__name__)

class FeastMetadataKeys(Enum):

LAST_UPDATED_TIMESTAMP = "last_updated_timestamp"

PROJECT_UUID = "project_uuid"

class SnowflakeRegistryConfig(RegistryConfig):

"""Registry config for Snowflake"""

registry_type: Literal["snowflake.registry"] = "snowflake.registry"

""" Registry type selector """

type: Literal["snowflake.registry"] = "snowflake.registry"

""" Registry type selector """

config_path: Optional[str] = os.path.expanduser("~/.snowsql/config")

""" Snowflake snowsql config path -- absolute path required (Cant use ~)"""

connection_name: Optional[str] = None

""" Snowflake connector connection name -- typically defined in ~/.snowflake/connections.toml """

account: Optional[str] = None

""" Snowflake deployment identifier -- drop .snowflakecomputing.com """

user: Optional[str] = None

""" Snowflake user name """

password: Optional[str] = None

""" Snowflake password """

role: Optional[str] = None

""" Snowflake role name """

warehouse: Optional[str] = None

""" Snowflake warehouse name """

authenticator: Optional[str] = None

""" Snowflake authenticator name """

private_key: Optional[str] = None

""" Snowflake private key file path"""

private_key_content: Optional[bytes] = None

""" Snowflake private key stored as bytes"""

private_key_passphrase: Optional[str] = None

""" Snowflake private key file passphrase"""

database: StrictStr

""" Snowflake database name """

schema_: Optional[str] = Field("PUBLIC", alias="schema")

""" Snowflake schema name """

model_config = ConfigDict(populate_by_name=True, extra="allow")

class SnowflakeRegistry(BaseRegistry):

def __init__(

self,

registry_config,

project: str,

repo_path,

):

assert registry_config is not None and isinstance(

registry_config, SnowflakeRegistryConfig

), "SnowflakeRegistry needs a valid registry_config, a path does not work"

self.registry_config = registry_config

self.registry_path = (

f'"{self.registry_config.database}"."{self.registry_config.schema_}"'

)

with GetSnowflakeConnection(self.registry_config) as conn:

sql_function_file = f"{os.path.dirname(feast.__file__)}/infra/utils/snowflake/registry/snowflake_table_creation.sql"

with open(sql_function_file, "r") as file:

sql_cmds = [

cmd.strip() for cmd in file.read().split(";") if cmd.strip()

]

for command in sql_cmds:

query = command.replace("REGISTRY_PATH", f"{self.registry_path}")

execute_snowflake_statement(conn, query)

self.purge_feast_metadata = registry_config.purge_feast_metadata

self.serve_features_while_materializing = (

registry_config.serve_features_while_materializing

)

self.project = project

# Initialize cache state before any method that may trigger

# _refresh_cached_registry_if_necessary (e.g. proto(), get_project()).

self._refresh_lock = Lock()

self.cached_registry_proto = None

self.cached_registry_proto_created = None

self.cached_registry_proto_ttl = timedelta(

seconds=(

registry_config.cache_ttl_seconds

if registry_config.cache_ttl_seconds is not None

else 0

)

)

self._sync_feast_metadata_to_projects_table()

if not self.purge_feast_metadata:

self._maybe_init_project_metadata(project)

self.cached_registry_proto = self.proto()

self.cached_registry_proto_created = _utc_now()

def _sync_feast_metadata_to_projects_table(self):

feast_metadata_projects: set[str] = set()

projects_set: set[str] = set()

with GetSnowflakeConnection(self.registry_config) as conn:

query = (

f'SELECT DISTINCT project_id FROM {self.registry_path}."FEAST_METADATA"'

)

df = execute_snowflake_statement(conn, query).fetch_pandas_all()

for row in df.iterrows():

feast_metadata_projects.add(row[1]["PROJECT_ID"])

if len(feast_metadata_projects) > 0:

with GetSnowflakeConnection(self.registry_config) as conn:

query = f'SELECT project_id FROM {self.registry_path}."PROJECTS"'

df = execute_snowflake_statement(conn, query).fetch_pandas_all()

for row in df.iterrows():

projects_set.add(row[1]["PROJECT_ID"])

# Find object in feast_metadata_projects but not in projects

projects_to_sync = feast_metadata_projects - projects_set

for project_name in projects_to_sync:

self.apply_project(Project(name=project_name), commit=True)

if self.purge_feast_metadata:

with GetSnowflakeConnection(self.registry_config) as conn:

query = f"""

DELETE FROM {self.registry_path}."FEAST_METADATA"

"""

execute_snowflake_statement(conn, query)

def refresh(self, project: Optional[str] = None):

self.cached_registry_proto = self.proto()

self.cached_registry_proto_created = _utc_now()

def _refresh_cached_registry_if_necessary(self) -> RegistryProto:

with self._refresh_lock:

expired = (

self.cached_registry_proto is None

or self.cached_registry_proto_created is None

) or (

self.cached_registry_proto_ttl.total_seconds()

> 0 # 0 ttl means infinity

and (

_utc_now()

> (

self.cached_registry_proto_created

+ self.cached_registry_proto_ttl

)

)

)

if expired:

logger.info("Registry cache expired, so refreshing")

self.refresh()

if self.cached_registry_proto is None:

raise RuntimeError("Registry cache is unexpectedly empty after refresh")

return self.cached_registry_proto

def teardown(self):

with GetSnowflakeConnection(self.registry_config) as conn:

sql_function_file = f"{os.path.dirname(feast.__file__)}/infra/utils/snowflake/registry/snowflake_table_deletion.sql"

with open(sql_function_file, "r") as file:

sql_cmds = [

cmd.strip() for cmd in file.read().split(";") if cmd.strip()

]

for command in sql_cmds:

query = command.replace("REGISTRY_PATH", f"{self.registry_path}")

execute_snowflake_statement(conn, query)

# apply operations

def apply_data_source(

self, data_source: DataSource, project: str, commit: bool = True

):

return self._apply_object(

"DATA_SOURCES",

project,

"DATA_SOURCE_NAME",

data_source,

"DATA_SOURCE_PROTO",

)

def apply_entity(self, entity: Entity, project: str, commit: bool = True):

return self._apply_object(

"ENTITIES", project, "ENTITY_NAME", entity, "ENTITY_PROTO"

)

def apply_feature_service(

self, feature_service: FeatureService, project: str, commit: bool = True

):

feature_service.prepare_for_apply(self, project, allow_cache=True)

return self._apply_object(

"FEATURE_SERVICES",

project,

"FEATURE_SERVICE_NAME",

feature_service,

"FEATURE_SERVICE_PROTO",

)

def apply_feature_view(

self,

feature_view: BaseFeatureView,

project: str,

commit: bool = True,

no_promote: bool = False,

):

if no_promote:

raise NotImplementedError(

"Feature view versioning (no_promote) is not supported by the Snowflake registry. "

"Use the SQL registry or file registry for versioning support."

)

feature_view.ensure_valid()

fv_table_str = self._infer_fv_table(feature_view)

fv_column_name = fv_table_str[:-1]

return self._apply_object(

fv_table_str,

project,

f"{fv_column_name}_NAME",

feature_view,

f"{fv_column_name}_PROTO",

)

def apply_saved_dataset(

self,

saved_dataset: SavedDataset,

project: str,

commit: bool = True,

):

return self._apply_object(

"SAVED_DATASETS",

project,

"SAVED_DATASET_NAME",

saved_dataset,

"SAVED_DATASET_PROTO",

)

def apply_validation_reference(

self,

validation_reference: ValidationReference,

project: str,

commit: bool = True,

):

return self._apply_object(

"VALIDATION_REFERENCES",

project,

"VALIDATION_REFERENCE_NAME",

validation_reference,

"VALIDATION_REFERENCE_PROTO",

)

def update_infra(self, infra: Infra, project: str, commit: bool = True):

self._apply_object(

"MANAGED_INFRA",

project,

"INFRA_NAME",

infra,

"INFRA_PROTO",

name="infra_obj",

)

def _initialize_project_if_not_exists(self, project_name: str):

try:

self.get_project(project_name, allow_cache=True)

return

except ProjectObjectNotFoundException:

try:

self.get_project(project_name, allow_cache=False)

return

except ProjectObjectNotFoundException:

self.apply_project(Project(name=project_name), commit=True)

def _apply_object(

self,

table: str,

project: str,

id_field_name: str,

obj: Any,

proto_field_name: str,

name: Optional[str] = None,

):

if not self.purge_feast_metadata:

self._maybe_init_project_metadata(project)

# Initialize project is necessary because FeatureStore object can apply objects individually without "feast apply" cli option

if not isinstance(obj, Project):

self._initialize_project_if_not_exists(project_name=project)

name = name or (obj.name if hasattr(obj, "name") else None)

assert name, f"name needs to be provided for {obj}"

update_datetime = _utc_now()

if hasattr(obj, "last_updated_timestamp"):

obj.last_updated_timestamp = update_datetime

with GetSnowflakeConnection(self.registry_config) as conn:

query = f"""

SELECT

project_id

FROM

{self.registry_path}."{table}"

WHERE

project_id = '{project}'

AND {id_field_name.lower()} = '{name}'

LIMIT 1

"""

df = execute_snowflake_statement(conn, query).fetch_pandas_all()

if not df.empty:

proto = hexlify(obj.to_proto().SerializeToString()).__str__()[1:]

query = f"""

UPDATE {self.registry_path}."{table}"

SET

{proto_field_name} = TO_BINARY({proto}),

last_updated_timestamp = CURRENT_TIMESTAMP()

WHERE

project_id = '{project}'

AND {id_field_name.lower()} = '{name}'

"""

execute_snowflake_statement(conn, query)

else:

obj_proto = obj.to_proto()

if hasattr(obj_proto, "meta") and hasattr(

obj_proto.meta, "created_timestamp"

):

obj_proto.meta.created_timestamp.FromDatetime(update_datetime)

proto = hexlify(obj_proto.SerializeToString()).__str__()[1:]

if table == "FEATURE_VIEWS":

query = f"""

INSERT INTO {self.registry_path}."{table}"

VALUES

('{name}', '{project}', CURRENT_TIMESTAMP(), TO_BINARY({proto}), '', '')

"""

elif "_FEATURE_VIEWS" in table or table == "LABEL_VIEWS":

query = f"""

INSERT INTO {self.registry_path}."{table}"

VALUES

('{name}', '{project}', CURRENT_TIMESTAMP(), TO_BINARY({proto}), '')

"""

else:

query = f"""

INSERT INTO {self.registry_path}."{table}"

VALUES

('{name}', '{project}', CURRENT_TIMESTAMP(), TO_BINARY({proto}))

"""

execute_snowflake_statement(conn, query)

if not isinstance(obj, Project):

self.apply_project(

self.get_project(name=project, allow_cache=False), commit=True

)

if not self.purge_feast_metadata:

self._set_last_updated_metadata(update_datetime, project)

def apply_permission(

self, permission: Permission, project: str, commit: bool = True

):

return self._apply_object(

"PERMISSIONS",

project,

"PERMISSION_NAME",

permission,

"PERMISSION_PROTO",

)

# delete operations

def delete_data_source(self, name: str, project: str, commit: bool = True):

return self._delete_object(

"DATA_SOURCES",

name,

project,

"DATA_SOURCE_NAME",

DataSourceObjectNotFoundException,

)

def delete_entity(self, name: str, project: str, commit: bool = True):

return self._delete_object(

"ENTITIES", name, project, "ENTITY_NAME", EntityNotFoundException

)

def delete_feature_service(self, name: str, project: str, commit: bool = True):

return self._delete_object(

"FEATURE_SERVICES",

name,

project,

"FEATURE_SERVICE_NAME",

FeatureServiceNotFoundException,

)

def delete_feature_view(self, name: str, project: str, commit: bool = True):

deleted_count = 0

_FV_TABLE_ID_COLUMNS = {

"FEATURE_VIEWS": "FEATURE_VIEW_NAME",

"ON_DEMAND_FEATURE_VIEWS": "FEATURE_VIEW_NAME",

"STREAM_FEATURE_VIEWS": "FEATURE_VIEW_NAME",

"LABEL_VIEWS": "LABEL_VIEW_NAME",

}

for table, id_col in _FV_TABLE_ID_COLUMNS.items():

deleted_count += self._delete_object(table, name, project, id_col, None)

if deleted_count == 0:

raise FeatureViewNotFoundException(name, project)

def delete_saved_dataset(self, name: str, project: str, allow_cache: bool = False):

self._delete_object(

"SAVED_DATASETS",

name,

project,

"SAVED_DATASET_NAME",

SavedDatasetNotFound,

)

def delete_validation_reference(self, name: str, project: str, commit: bool = True):

self._delete_object(

"VALIDATION_REFERENCES",

name,

project,

"VALIDATION_REFERENCE_NAME",

ValidationReferenceNotFound,

)

def _delete_object(

self,

table: str,

name: str,

project: str,

id_field_name: str,

not_found_exception: Optional[Callable],

):

with GetSnowflakeConnection(self.registry_config) as conn:

query = f"""

DELETE FROM {self.registry_path}."{table}"

WHERE

project_id = '{project}'

AND {id_field_name.lower()} = '{name}'

"""

cursor = execute_snowflake_statement(conn, query)

if cursor.rowcount < 1 and not_found_exception: # type: ignore

raise not_found_exception(name, project)

self._set_last_updated_metadata(_utc_now(), project)

return cursor.rowcount

def delete_permission(self, name: str, project: str, commit: bool = True):

return self._delete_object(

"PERMISSIONS",

name,

project,

"PERMISSION_NAME",

PermissionNotFoundException,

)

# get operations

def get_data_source(

self, name: str, project: str, allow_cache: bool = False

) -> DataSource:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_data_source(registry_proto, name, project)

return self._get_object(

"DATA_SOURCES",

name,

project,

DataSourceProto,

DataSource,

"DATA_SOURCE_NAME",

"DATA_SOURCE_PROTO",

DataSourceObjectNotFoundException,

)

def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Entity:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_entity(registry_proto, name, project)

return self._get_object(

"ENTITIES",

name,

project,

EntityProto,

Entity,

"ENTITY_NAME",

"ENTITY_PROTO",

EntityNotFoundException,

)

def get_feature_service(

self, name: str, project: str, allow_cache: bool = False

) -> FeatureService:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_feature_service(

registry_proto, name, project

)

return self._get_object(

"FEATURE_SERVICES",

name,

project,

FeatureServiceProto,

FeatureService,

"FEATURE_SERVICE_NAME",

"FEATURE_SERVICE_PROTO",

FeatureServiceNotFoundException,

)

def get_feature_view(

self, name: str, project: str, allow_cache: bool = False

) -> FeatureView:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_feature_view(registry_proto, name, project)

return self._get_object(

"FEATURE_VIEWS",

name,

project,

FeatureViewProto,

FeatureView,

"FEATURE_VIEW_NAME",

"FEATURE_VIEW_PROTO",

FeatureViewNotFoundException,

)

def get_any_feature_view(

self, name: str, project: str, allow_cache: bool = False

) -> BaseFeatureView:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_any_feature_view(

registry_proto, name, project

)

fv = self._get_object(

"FEATURE_VIEWS",

name,

project,

FeatureViewProto,

FeatureView,

"FEATURE_VIEW_NAME",

"FEATURE_VIEW_PROTO",

None,

)

if not fv:

fv = self._get_object(

"STREAM_FEATURE_VIEWS",

name,

project,

StreamFeatureViewProto,

StreamFeatureView,

"STREAM_FEATURE_VIEW_NAME",

"STREAM_FEATURE_VIEW_PROTO",

None,

)

if not fv:

fv = self._get_object(

"ON_DEMAND_FEATURE_VIEWS",

name,

project,

OnDemandFeatureViewProto,

OnDemandFeatureView,

"ON_DEMAND_FEATURE_VIEW_NAME",

"ON_DEMAND_FEATURE_VIEW_PROTO",

None,

)

if not fv:

fv = self._get_object(

"LABEL_VIEWS",

name,

project,

LabelViewProto,

LabelView,

"LABEL_VIEW_NAME",

"LABEL_VIEW_PROTO",

FeatureViewNotFoundException,

)

return fv

def list_all_feature_views(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

skip_udf: bool = False,

updated_since: Optional[datetime] = None,

) -> List[BaseFeatureView]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

feature_views = proto_registry_utils.list_all_feature_views(

registry_proto, project, tags, skip_udf=skip_udf

)

if updated_since is not None:

cutoff = to_naive_utc(updated_since)

feature_views = [

fv

for fv in feature_views

if fv.last_updated_timestamp is not None

and fv.last_updated_timestamp >= cutoff

]

return feature_views

feature_views = (

cast(

list[BaseFeatureView],

self.list_feature_views(project, allow_cache, tags, skip_udf=skip_udf),

)

+ cast(

list[BaseFeatureView],

self.list_stream_feature_views(

project, allow_cache, tags, skip_udf=skip_udf

),

)

+ cast(

list[BaseFeatureView],

self.list_on_demand_feature_views(

project, allow_cache, tags, skip_udf=skip_udf

),

)

+ cast(

list[BaseFeatureView],

self.list_label_views(project, allow_cache, tags),

)

)

if updated_since is not None:

cutoff = to_naive_utc(updated_since)

feature_views = [

fv

for fv in feature_views

if fv.last_updated_timestamp is not None

and fv.last_updated_timestamp >= cutoff

]

return feature_views

def get_infra(self, project: str, allow_cache: bool = False) -> Infra:

infra_object = self._get_object(

"MANAGED_INFRA",

"infra_obj",

project,

InfraProto,

Infra,

"INFRA_NAME",

"INFRA_PROTO",

None,

)

infra_object = infra_object or InfraProto()

return Infra.from_proto(infra_object)

def get_on_demand_feature_view(

self, name: str, project: str, allow_cache: bool = False

) -> OnDemandFeatureView:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_on_demand_feature_view(

registry_proto, name, project

)

return self._get_object(

"ON_DEMAND_FEATURE_VIEWS",

name,

project,

OnDemandFeatureViewProto,

OnDemandFeatureView,

"ON_DEMAND_FEATURE_VIEW_NAME",

"ON_DEMAND_FEATURE_VIEW_PROTO",

FeatureViewNotFoundException,

)

def get_saved_dataset(

self, name: str, project: str, allow_cache: bool = False

) -> SavedDataset:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_saved_dataset(registry_proto, name, project)

return self._get_object(

"SAVED_DATASETS",

name,

project,

SavedDatasetProto,

SavedDataset,

"SAVED_DATASET_NAME",

"SAVED_DATASET_PROTO",

SavedDatasetNotFound,

)

def get_stream_feature_view(

self, name: str, project: str, allow_cache: bool = False

):

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_stream_feature_view(

registry_proto, name, project

)

return self._get_object(

"STREAM_FEATURE_VIEWS",

name,

project,

StreamFeatureViewProto,

StreamFeatureView,

"STREAM_FEATURE_VIEW_NAME",

"STREAM_FEATURE_VIEW_PROTO",

FeatureViewNotFoundException,

)

def get_validation_reference(

self, name: str, project: str, allow_cache: bool = False

) -> ValidationReference:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_validation_reference(

registry_proto, name, project

)

return self._get_object(

"VALIDATION_REFERENCES",

name,

project,

ValidationReferenceProto,

ValidationReference,

"VALIDATION_REFERENCE_NAME",

"VALIDATION_REFERENCE_PROTO",

ValidationReferenceNotFound,

)

def _get_object(

self,

table: str,

name: str,

project: str,

proto_class: Any,

python_class: Any,

id_field_name: str,

proto_field_name: str,

not_found_exception: Optional[Callable],

):

with GetSnowflakeConnection(self.registry_config) as conn:

query = f"""

SELECT

{proto_field_name}

FROM

{self.registry_path}."{table}"

WHERE

project_id = '{project}'

AND {id_field_name.lower()} = '{name}'

LIMIT 1

"""

df = execute_snowflake_statement(conn, query).fetch_pandas_all()

if not df.empty:

_proto = proto_class.FromString(df.squeeze())

return python_class.from_proto(_proto)

elif not_found_exception:

raise not_found_exception(name, project)

else:

return None

def get_permission(

self, name: str, project: str, allow_cache: bool = False

) -> Permission:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.get_permission(registry_proto, name, project)

return self._get_object(

"PERMISSIONS",

name,

project,

PermissionProto,

Permission,

"PERMISSION_NAME",

"PERMISSION_PROTO",

PermissionNotFoundException,

)

# list operations

def list_data_sources(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

) -> List[DataSource]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.list_data_sources(registry_proto, project, tags)

return self._list_objects(

"DATA_SOURCES",

project,

DataSourceProto,

DataSource,

"DATA_SOURCE_PROTO",

tags=tags,

)

def list_entities(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

) -> List[Entity]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.list_entities(registry_proto, project, tags)

return self._list_objects(

"ENTITIES", project, EntityProto, Entity, "ENTITY_PROTO", tags=tags

)

def list_feature_services(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

) -> List[FeatureService]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.list_feature_services(

registry_proto, project, tags

)

return self._list_objects(

"FEATURE_SERVICES",

project,

FeatureServiceProto,

FeatureService,

"FEATURE_SERVICE_PROTO",

tags=tags,

)

def list_feature_views(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

skip_udf: bool = False,

) -> List[FeatureView]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.list_feature_views(

registry_proto, project, tags, skip_udf=skip_udf

)

return self._list_objects(

"FEATURE_VIEWS",

project,

FeatureViewProto,

FeatureView,

"FEATURE_VIEW_PROTO",

tags=tags,

skip_udf=skip_udf,

)

def list_on_demand_feature_views(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

skip_udf: bool = False,

) -> List[OnDemandFeatureView]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.list_on_demand_feature_views(

registry_proto, project, tags, skip_udf=skip_udf

)

return self._list_objects(

"ON_DEMAND_FEATURE_VIEWS",

project,

OnDemandFeatureViewProto,

OnDemandFeatureView,

"ON_DEMAND_FEATURE_VIEW_PROTO",

tags=tags,

skip_udf=skip_udf,

)

def list_saved_datasets(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

namespace: Optional[str] = None,

collection: Optional[str] = None,

) -> List[SavedDataset]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.list_saved_datasets(

registry_proto,

project,

tags,

namespace=namespace,

collection=collection,

)

results = self._list_objects(

"SAVED_DATASETS",

project,

SavedDatasetProto,

SavedDataset,

"SAVED_DATASET_PROTO",

tags=tags,

)

if namespace is not None:

results = [sd for sd in results if sd.namespace == namespace]

if collection is not None:

results = [sd for sd in results if sd.collection == collection]

return results

def list_stream_feature_views(

self,

project: str,

allow_cache: bool = False,

tags: Optional[dict[str, str]] = None,

skip_udf: bool = False,

) -> List[StreamFeatureView]:

if allow_cache:

registry_proto = self._refresh_cached_registry_if_necessary()

return proto_registry_utils.list_stream_feature_views(

registry_proto, project, tags, skip_udf=skip_udf

)

return self._list_objects(

"STREAM_FEATURE_VIEWS",

project,

StreamFeatureViewProto,

StreamFeatureView,

"STREAM_FEATURE_VIEW_PROTO",

skip_udf=skip_udf,

tags=tags,

)

def get_label_view(