import logging

from concurrent import futures

from datetime import datetime, timezone

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

import grpc

from google.protobuf.empty_pb2 import Empty

from grpc_health.v1 import health, health_pb2, health_pb2_grpc

from grpc_reflection.v1alpha import reflection

from feast import FeatureService, FeatureStore

from feast.base_feature_view import BaseFeatureView

from feast.data_source import DataSource

from feast.entity import Entity

from feast.errors import FeastObjectNotFoundException, FeastPermissionError

from feast.feast_object import FeastObject

from feast.feature_view import FeatureView

from feast.grpc_error_interceptor import ErrorInterceptor

from feast.infra.infra_object import Infra

from feast.infra.registry.base_registry import BaseRegistry

from feast.on_demand_feature_view import OnDemandFeatureView

from feast.permissions.action import AuthzedAction

from feast.permissions.permission import Permission

from feast.permissions.security_manager import (

assert_permissions,

assert_permissions_to_update,

get_security_manager,

is_auth_necessary,

permitted_resources,

)

from feast.permissions.server.grpc import AuthInterceptor

from feast.permissions.server.utils import (

AuthManagerType,

ServerType,

init_auth_manager,

init_security_manager,

str_to_auth_manager_type,

)

from feast.project import Project

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

from feast.protos.feast.registry import RegistryServer_pb2, RegistryServer_pb2_grpc

from feast.protos.feast.registry.RegistryServer_pb2 import Feature, ListFeaturesResponse

from feast.saved_dataset import SavedDataset, ValidationReference

from feast.stream_feature_view import StreamFeatureView

logger = logging.getLogger(__name__)

logger.setLevel(logging.INFO)

def apply_pagination_and_sorting(

objects: List[Any],

pagination: Optional[RegistryServer_pb2.PaginationParams] = None,

sorting: Optional[RegistryServer_pb2.SortingParams] = None,

) -> tuple[List[Any], RegistryServer_pb2.PaginationMetadata]:

"""

Apply sorting and pagination to a list of objects at the gRPC layer.

Args:

objects: List of objects to paginate and sort

pagination: Pagination parameters (page, limit)

sorting: Sorting parameters (sort_by, sort_order)

Returns:

Tuple of (paginated and sorted objects, pagination metadata)

"""

if not objects:

empty_metadata = RegistryServer_pb2.PaginationMetadata(

page=0,

limit=0,

total_count=0,

total_pages=0,

has_next=False,

has_previous=False,

)

return objects, empty_metadata

if sorting and sorting.sort_by and sorting.sort_by.strip():

objects = apply_sorting(objects, sorting.sort_by, sorting.sort_order)

total_count = len(objects)

# Apply pagination if requested

if pagination and pagination.page > 0 and pagination.limit > 0:

start_idx = (pagination.page - 1) * pagination.limit

end_idx = start_idx + pagination.limit

paginated_objects = objects[start_idx:end_idx]

total_pages = (total_count + pagination.limit - 1) // pagination.limit

has_next = pagination.page < total_pages

has_previous = pagination.page > 1

pagination_metadata = RegistryServer_pb2.PaginationMetadata(

page=pagination.page,

limit=pagination.limit,

total_count=total_count,

total_pages=total_pages,

has_next=has_next,

has_previous=has_previous,

)

return paginated_objects, pagination_metadata

else:

# No pagination requested - return all objects

pagination_metadata = RegistryServer_pb2.PaginationMetadata(

page=0,

limit=0,

total_count=total_count,

total_pages=1,

has_next=False,

has_previous=False,

)

return objects, pagination_metadata

def apply_sorting(objects: List[Any], sort_by: str, sort_order: str) -> List[Any]:

"""Apply sorting to a list of objects using dot notation for nested attributes."""

def get_nested_attr(obj, attr_path: str):

"""Get nested attribute using dot notation."""

attrs = attr_path.split(".")

current = obj

for attr in attrs:

if hasattr(current, attr):

current = getattr(current, attr)

elif isinstance(current, dict) and attr in current:

current = current[attr]

else:

return None

return current

def sort_key(obj):

value = get_nested_attr(obj, sort_by)

if value is None:

return ("", 0, None)

if isinstance(value, str):

return (value.lower(), 1, None)

elif isinstance(value, (int, float)):

return ("", 1, value)

else:

return (str(value).lower(), 1, None)

reverse = sort_order.lower() == "desc"

try:

return sorted(objects, key=sort_key, reverse=reverse)

except Exception as e:

logger.warning(f"Failed to sort objects by '{sort_by}': {e}")

return objects

def _build_any_feature_view_proto(feature_view: BaseFeatureView):

from feast.labeling.label_view import LabelView

if isinstance(feature_view, StreamFeatureView):

arg_name = "stream_feature_view"

feature_view_proto = feature_view.to_proto()

elif isinstance(feature_view, FeatureView):

arg_name = "feature_view"

feature_view_proto = feature_view.to_proto()

elif isinstance(feature_view, OnDemandFeatureView):

arg_name = "on_demand_feature_view"

feature_view_proto = feature_view.to_proto()

elif isinstance(feature_view, LabelView):

arg_name = "label_view"

feature_view_proto = feature_view.to_proto()

else:

raise ValueError(f"Unexpected feature view type: {type(feature_view)}")

return RegistryServer_pb2.AnyFeatureView(

feature_view=feature_view_proto if arg_name == "feature_view" else None,

stream_feature_view=feature_view_proto

if arg_name == "stream_feature_view"

else None,

on_demand_feature_view=feature_view_proto

if arg_name == "on_demand_feature_view"

else None,

label_view=feature_view_proto if arg_name == "label_view" else None,

)

class RegistryServer(RegistryServer_pb2_grpc.RegistryServerServicer):

def __init__(self, registry: BaseRegistry, store=None) -> None:

super().__init__()

self.proxied_registry = registry

self.store = store

_JOB_NAME_TEMPLATES: dict = {

"featureView": ["feast_apply_feature_view_{name}"],

"featureService": [

"feature_service_{name}",

"feast_apply_feature_service_{name}",

],

"savedDataset": ["saved_dataset_{name}"],

"onDemandFeatureView": ["feast_apply_odfv_{name}"],

}

@property

def _openlineage_enabled(self) -> bool:

"""Fast cached check: is OpenLineage configured and enabled?

Evaluated once per RegistryServer lifetime so Apply/Delete handlers

pay zero cost when OL is disabled.

"""

cached = getattr(self, "_ol_enabled_cache", None)

if cached is not None:

return cached

enabled = False

try:

if self.store:

ol_cfg = getattr(self.store.config, "openlineage", None)

if ol_cfg is not None and getattr(ol_cfg, "enabled", False):

enabled = True

except Exception:

pass

self._ol_enabled_cache = enabled

return enabled

def _emit_openlineage_for_objects(self, objects: list, project: str):

"""Emit OpenLineage events for objects modified via API/gRPC.

Skips entirely when OL is disabled — no imports, no emitter access.

All errors are caught so registry operations never fail due to OL.

"""

if not self._openlineage_enabled or not objects:

return

try:

emitter = self.store.openlineage_emitter

if emitter is not None:

emitter.emit_apply(objects, project)

except Exception as e:

logger.warning(f"Failed to emit OpenLineage events for API apply: {e}")

def _delete_openlineage_for_object(self, name: str, project: str, object_type: str):

"""Remove OpenLineage data for a deleted Feast object.

Skips entirely when OL consumer is not active.

All errors are caught so registry deletes never fail due to OL.

"""

if not self._openlineage_enabled:

return

try:

import feast.api.registry.rest as rest_module

ol_store = getattr(rest_module, "_ol_store_instance", None)

if ol_store is None:

return

ol_config = getattr(rest_module, "_ol_config", None)

namespace = (

ol_config.namespace if ol_config and ol_config.namespace else project

)

ol_store.delete_dataset(namespace, name)

for tpl in self._JOB_NAME_TEMPLATES.get(object_type, []):

ol_store.delete_job(namespace, tpl.format(name=name))

except Exception as e:

logger.warning(

f"Failed to delete OpenLineage data for {object_type}/{name}: {e}"

)

def Proto(self, request: Empty, context) -> RegistryProto:

"""Build a RegistryProto from individually RBAC-filtered list calls.

The ``RegistryServer.Proto`` RPC must honor the same permission checks as the

other RPCs rather than returning ``proxied_registry.proto()`` directly, which

would bypass RBAC and expose every object (entities, feature views, data

sources, permissions, projects, etc.) regardless of authorization.

Each object type is filtered with ``permitted_resources(..., DESCRIBE)``: under

``NoAuthConfig`` this is a no-op (the full registry is returned, so remote

registries keep working), while with auth enabled the caller only sees the

objects they are permitted to ``DESCRIBE``.

"""

def describable(resources: list) -> list:

return permitted_resources(

resources=cast(list[FeastObject], resources),

actions=AuthzedAction.DESCRIBE,

)

registry_proto = RegistryProto()

for project in describable(self.proxied_registry.list_projects()):

registry_proto.projects.append(project.to_proto())

project_name = project.name

for entity in describable(

self.proxied_registry.list_entities(project=project_name)

):

registry_proto.entities.append(entity.to_proto())

for data_source in describable(

self.proxied_registry.list_data_sources(project=project_name)

):

registry_proto.data_sources.append(data_source.to_proto())

for feature_view in describable(

self.proxied_registry.list_feature_views(project=project_name)

):

registry_proto.feature_views.append(feature_view.to_proto())

for stream_feature_view in describable(

self.proxied_registry.list_stream_feature_views(project=project_name)

):

registry_proto.stream_feature_views.append(

stream_feature_view.to_proto()

)

for on_demand_feature_view in describable(

self.proxied_registry.list_on_demand_feature_views(project=project_name)

):

registry_proto.on_demand_feature_views.append(

on_demand_feature_view.to_proto()

)

for label_view in describable(

self.proxied_registry.list_label_views(project=project_name)

):

registry_proto.label_views.append(label_view.to_proto())

for feature_service in describable(

self.proxied_registry.list_feature_services(project=project_name)

):

registry_proto.feature_services.append(feature_service.to_proto())

for saved_dataset in describable(

self.proxied_registry.list_saved_datasets(project=project_name)

):

registry_proto.saved_datasets.append(saved_dataset.to_proto())

for validation_reference in describable(

self.proxied_registry.list_validation_references(project=project_name)

):

registry_proto.validation_references.append(

validation_reference.to_proto()

)

for permission in describable(

self.proxied_registry.list_permissions(project=project_name)

):

registry_proto.permissions.append(permission.to_proto())

# Carry the registry's real last_updated/version_id rather than stamping "now":

# this proto is rebuilt from individual list calls (for RBAC filtering), but it must

# not look like a fresh commit on every call — clients such as the remote feature

# server key cache freshness off this metadata. Reading these two scalar fields from

# the source proto leaks nothing RBAC-protected (no objects are copied from it).

source_proto = self.proxied_registry.proto()

registry_proto.last_updated.CopyFrom(source_proto.last_updated)

registry_proto.version_id = source_proto.version_id

return registry_proto

def ApplyEntity(self, request: RegistryServer_pb2.ApplyEntityRequest, context):

entity = cast(

Entity,

assert_permissions_to_update(

resource=Entity.from_proto(request.entity),

getter=self.proxied_registry.get_entity,

project=request.project,

),

)

self.proxied_registry.apply_entity(

entity=entity,

project=request.project,

commit=request.commit,

)

self._emit_openlineage_for_objects([entity], request.project)

return Empty()

def GetEntity(self, request: RegistryServer_pb2.GetEntityRequest, context):

return assert_permissions(

self.proxied_registry.get_entity(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

actions=[AuthzedAction.DESCRIBE],

).to_proto()

def ListEntities(self, request: RegistryServer_pb2.ListEntitiesRequest, context):

paginated_entities, pagination_metadata = apply_pagination_and_sorting(

permitted_resources(

resources=cast(

list[FeastObject],

self.proxied_registry.list_entities(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

),

),

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

return RegistryServer_pb2.ListEntitiesResponse(

entities=[entity.to_proto() for entity in paginated_entities],

pagination=pagination_metadata,

)

def DeleteEntity(self, request: RegistryServer_pb2.DeleteEntityRequest, context):

assert_permissions(

resource=self.proxied_registry.get_entity(

name=request.name, project=request.project

),

actions=AuthzedAction.DELETE,

)

self.proxied_registry.delete_entity(

name=request.name, project=request.project, commit=request.commit

)

self._delete_openlineage_for_object(request.name, request.project, "entity")

return Empty()

def ApplyDataSource(

self, request: RegistryServer_pb2.ApplyDataSourceRequest, context

):

data_source = cast(

DataSource,

assert_permissions_to_update(

resource=DataSource.from_proto(request.data_source),

getter=self.proxied_registry.get_data_source,

project=request.project,

),

)

self.proxied_registry.apply_data_source(

data_source=data_source,

project=request.project,

commit=request.commit,

)

self._emit_openlineage_for_objects([data_source], request.project)

return Empty()

def GetDataSource(self, request: RegistryServer_pb2.GetDataSourceRequest, context):

return assert_permissions(

resource=self.proxied_registry.get_data_source(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

actions=AuthzedAction.DESCRIBE,

).to_proto()

def ListDataSources(

self, request: RegistryServer_pb2.ListDataSourcesRequest, context

):

paginated_data_sources, pagination_metadata = apply_pagination_and_sorting(

permitted_resources(

resources=cast(

list[FeastObject],

self.proxied_registry.list_data_sources(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

),

),

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

return RegistryServer_pb2.ListDataSourcesResponse(

data_sources=[

data_source.to_proto() for data_source in paginated_data_sources

],

pagination=pagination_metadata,

)

def DeleteDataSource(

self, request: RegistryServer_pb2.DeleteDataSourceRequest, context

):

assert_permissions(

resource=self.proxied_registry.get_data_source(

name=request.name,

project=request.project,

),

actions=AuthzedAction.DELETE,

)

self.proxied_registry.delete_data_source(

name=request.name, project=request.project, commit=request.commit

)

self._delete_openlineage_for_object(request.name, request.project, "dataSource")

return Empty()

def GetFeatureView(

self, request: RegistryServer_pb2.GetFeatureViewRequest, context

):

return assert_permissions(

self.proxied_registry.get_feature_view(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

actions=[AuthzedAction.DESCRIBE],

).to_proto()

def GetAnyFeatureView(

self, request: RegistryServer_pb2.GetAnyFeatureViewRequest, context

):

feature_view = assert_permissions(

cast(

FeastObject,

self.proxied_registry.get_any_feature_view(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

),

actions=[AuthzedAction.DESCRIBE],

)

return RegistryServer_pb2.GetAnyFeatureViewResponse(

any_feature_view=_build_any_feature_view_proto(

cast(BaseFeatureView, feature_view)

)

)

def ApplyFeatureView(

self, request: RegistryServer_pb2.ApplyFeatureViewRequest, context

):

from feast.labeling.label_view import LabelView

feature_view_type = request.WhichOneof("base_feature_view")

feature_view_meta: BaseFeatureView

if feature_view_type == "feature_view":

feature_view_meta = FeatureView.from_proto(

request.feature_view, skip_udf=True

)

elif feature_view_type == "on_demand_feature_view":

feature_view_meta = OnDemandFeatureView.from_proto(

request.on_demand_feature_view, skip_udf=True

)

elif feature_view_type == "stream_feature_view":

feature_view_meta = StreamFeatureView.from_proto(

request.stream_feature_view, skip_udf=True

)

elif feature_view_type == "label_view":

feature_view_meta = LabelView.from_proto(request.label_view)

else:

raise ValueError(f"Unexpected feature view type: {feature_view_type}")

getter: Callable

if isinstance(feature_view_meta, StreamFeatureView):

getter = self.proxied_registry.get_stream_feature_view

elif isinstance(feature_view_meta, FeatureView):

getter = self.proxied_registry.get_feature_view

elif isinstance(feature_view_meta, OnDemandFeatureView):

getter = self.proxied_registry.get_on_demand_feature_view

elif isinstance(feature_view_meta, LabelView):

getter = self.proxied_registry.get_label_view

else:

getter = self.proxied_registry.get_feature_view

assert_permissions_to_update(

resource=cast(FeastObject, feature_view_meta),

getter=cast(Callable, getter),

project=request.project,

)

feature_view: BaseFeatureView

if feature_view_type == "feature_view":

feature_view = FeatureView.from_proto(request.feature_view)

elif feature_view_type == "on_demand_feature_view":

feature_view = OnDemandFeatureView.from_proto(

request.on_demand_feature_view

)

elif feature_view_type == "stream_feature_view":

feature_view = StreamFeatureView.from_proto(request.stream_feature_view)

elif feature_view_type == "label_view":

feature_view = LabelView.from_proto(request.label_view)

else:

raise ValueError(f"Unexpected feature view type: {feature_view_type}")

self.proxied_registry.apply_feature_view(

feature_view=feature_view,

project=request.project,

commit=request.commit,

)

self._emit_openlineage_for_objects([feature_view], request.project)

return Empty()

def ListFeatureViews(

self, request: RegistryServer_pb2.ListFeatureViewsRequest, context

):

paginated_feature_views, pagination_metadata = apply_pagination_and_sorting(

permitted_resources(

resources=cast(

list[FeastObject],

self.proxied_registry.list_feature_views(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

skip_udf=True,

),

),

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

return RegistryServer_pb2.ListFeatureViewsResponse(

feature_views=[

feature_view.to_proto() for feature_view in paginated_feature_views

],

pagination=pagination_metadata,

)

def ListAllFeatureViews(

self, request: RegistryServer_pb2.ListAllFeatureViewsRequest, context

):

from feast.labeling.label_view import LabelView

updated_since = None

if request.HasField("updated_since"):

updated_since = request.updated_since.ToDatetime().replace(

tzinfo=timezone.utc

)

all_feature_views = cast(

list[FeastObject],

[

fv

for fv in self.proxied_registry.list_all_feature_views(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

skip_udf=True,

updated_since=updated_since,

)

if not isinstance(fv, LabelView)

],

)

if (

request.entity

or request.feature

or request.feature_service

or request.data_source

):

filtered_feature_views = []

for feature_view in all_feature_views:

feature_view_obj = cast(BaseFeatureView, feature_view)

include_feature_view = True

# Filter by entity

if request.entity:

entity_match = False

if (

hasattr(feature_view_obj, "entities")

and feature_view_obj.entities

):

if request.entity in feature_view_obj.entities:

entity_match = True

elif (

hasattr(feature_view_obj, "entity_columns")

and feature_view_obj.entity_columns

):

# For feature views with entity_columns, check if the entity is referenced

entity_names = [

col.entity_name

for col in feature_view_obj.entity_columns

if hasattr(col, "entity_name")

]

if request.entity in entity_names:

entity_match = True

if not entity_match:

include_feature_view = False

# Filter by feature

if include_feature_view and request.feature:

feature_match = False

if hasattr(feature_view_obj, "features"):

feature_names = [f.name for f in feature_view_obj.features]

if request.feature in feature_names:

feature_match = True

if not feature_match:

include_feature_view = False

# Filter by feature service

if include_feature_view and request.feature_service:

try:

feature_service = self.proxied_registry.get_feature_service(

name=request.feature_service,

project=request.project,

allow_cache=request.allow_cache,

)

assert_permissions(

resource=feature_service,

actions=AuthzedAction.DESCRIBE,

)

feature_service_match = False

if hasattr(feature_service, "feature_view_projections"):

for (

feature_view_ref

) in feature_service.feature_view_projections:

if feature_view_ref.name == feature_view_obj.name:

feature_service_match = True

break

if not feature_service_match:

include_feature_view = False

except Exception:

include_feature_view = False

# Filter by data source

if include_feature_view and request.data_source:

data_source_match = False

if (

hasattr(feature_view_obj, "batch_source")

and feature_view_obj.batch_source

):

if feature_view_obj.batch_source.name == request.data_source:

data_source_match = True

if (

hasattr(feature_view_obj, "stream_source")

and feature_view_obj.stream_source

):

if feature_view_obj.stream_source.name == request.data_source:

data_source_match = True

if (

hasattr(feature_view_obj, "source_request_sources")

and feature_view_obj.source_request_sources

):

for (

request_source

) in feature_view_obj.source_request_sources.values():

if request_source.name == request.data_source:

data_source_match = True

break

if not data_source_match:

include_feature_view = False

if include_feature_view:

filtered_feature_views.append(feature_view_obj)

all_feature_views = cast(list[FeastObject], filtered_feature_views)

paginated_feature_views, pagination_metadata = apply_pagination_and_sorting(

permitted_resources(

resources=all_feature_views,

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

return RegistryServer_pb2.ListAllFeatureViewsResponse(

feature_views=[

_build_any_feature_view_proto(cast(BaseFeatureView, feature_view))

for feature_view in paginated_feature_views

],

pagination=pagination_metadata,

)

def DeleteFeatureView(

self, request: RegistryServer_pb2.DeleteFeatureViewRequest, context

):

feature_view = self.proxied_registry.get_any_feature_view(

name=request.name, project=request.project, allow_cache=False

)

from feast.on_demand_feature_view import OnDemandFeatureView

fv_type = (

"onDemandFeatureView"

if isinstance(feature_view, OnDemandFeatureView)

else "featureView"

)

assert_permissions(

resource=cast(FeastObject, feature_view),

actions=[AuthzedAction.DELETE],

)

self.proxied_registry.delete_feature_view(

name=request.name, project=request.project, commit=request.commit

)

self._delete_openlineage_for_object(request.name, request.project, fv_type)

return Empty()

def GetStreamFeatureView(

self, request: RegistryServer_pb2.GetStreamFeatureViewRequest, context

):

return assert_permissions(

resource=self.proxied_registry.get_stream_feature_view(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

actions=[AuthzedAction.DESCRIBE],

).to_proto()

def ListStreamFeatureViews(

self, request: RegistryServer_pb2.ListStreamFeatureViewsRequest, context

):

paginated_stream_feature_views, pagination_metadata = (

apply_pagination_and_sorting(

permitted_resources(

resources=cast(

list[FeastObject],

self.proxied_registry.list_stream_feature_views(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

skip_udf=True,

),

),

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

)

return RegistryServer_pb2.ListStreamFeatureViewsResponse(

stream_feature_views=[

stream_feature_view.to_proto()

for stream_feature_view in paginated_stream_feature_views

],

pagination=pagination_metadata,

)

def GetOnDemandFeatureView(

self, request: RegistryServer_pb2.GetOnDemandFeatureViewRequest, context

):

return assert_permissions(

resource=self.proxied_registry.get_on_demand_feature_view(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

actions=[AuthzedAction.DESCRIBE],

).to_proto()

def ListOnDemandFeatureViews(

self, request: RegistryServer_pb2.ListOnDemandFeatureViewsRequest, context

):

paginated_on_demand_feature_views, pagination_metadata = (

apply_pagination_and_sorting(

permitted_resources(

resources=cast(

list[FeastObject],

self.proxied_registry.list_on_demand_feature_views(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

skip_udf=True,

),

),

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

)

return RegistryServer_pb2.ListOnDemandFeatureViewsResponse(

on_demand_feature_views=[

on_demand_feature_view.to_proto()

for on_demand_feature_view in paginated_on_demand_feature_views

],

pagination=pagination_metadata,

)

def GetLabelView(self, request: RegistryServer_pb2.GetLabelViewRequest, context):

return assert_permissions(

resource=self.proxied_registry.get_label_view(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

actions=[AuthzedAction.DESCRIBE],

).to_proto()

def ListLabelViews(

self, request: RegistryServer_pb2.ListLabelViewsRequest, context

):

paginated_label_views, pagination_metadata = apply_pagination_and_sorting(

permitted_resources(

resources=cast(

list[FeastObject],

self.proxied_registry.list_label_views(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

),

),

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

return RegistryServer_pb2.ListLabelViewsResponse(

label_views=[label_view.to_proto() for label_view in paginated_label_views],

pagination=pagination_metadata,

)

def ApplyFeatureService(

self, request: RegistryServer_pb2.ApplyFeatureServiceRequest, context

):

feature_service = cast(

FeatureService,

assert_permissions_to_update(

resource=FeatureService.from_proto(request.feature_service),

getter=self.proxied_registry.get_feature_service,

project=request.project,

),

)

self.proxied_registry.apply_feature_service(

feature_service=feature_service,

project=request.project,

commit=request.commit,

)

self._emit_openlineage_for_objects([feature_service], request.project)

return Empty()

def GetFeatureService(

self, request: RegistryServer_pb2.GetFeatureServiceRequest, context

):

return assert_permissions(

resource=self.proxied_registry.get_feature_service(

name=request.name,

project=request.project,

allow_cache=request.allow_cache,

),

actions=[AuthzedAction.DESCRIBE],

).to_proto()

def ListFeatureServices(

self, request: RegistryServer_pb2.ListFeatureServicesRequest, context

):

# Get all feature services first

all_feature_services = self.proxied_registry.list_feature_services(

project=request.project,

allow_cache=request.allow_cache,

tags=dict(request.tags),

)

# Filter by feature view if specified

if request.feature_view:

filtered_feature_services = []

for feature_service in all_feature_services:

# Check if the feature service uses the specified feature view

feature_view_match = False

if hasattr(feature_service, "feature_view_projections"):

for (

feature_view_projection

) in feature_service.feature_view_projections:

if feature_view_projection.name == request.feature_view:

feature_view_match = True

break

if feature_view_match:

filtered_feature_services.append(feature_service)

all_feature_services = filtered_feature_services

paginated_feature_services, pagination_metadata = apply_pagination_and_sorting(

permitted_resources(

resources=cast(list[FeastObject], all_feature_services),

actions=AuthzedAction.DESCRIBE,

),

pagination=request.pagination,

sorting=request.sorting,

)

return RegistryServer_pb2.ListFeatureServicesResponse(

feature_services=[

feature_service.to_proto()

for feature_service in paginated_feature_services

],

pagination=pagination_metadata,

)

def DeleteFeatureService(

self, request: RegistryServer_pb2.DeleteFeatureServiceRequest, context

):

feature_service = self.proxied_registry.get_feature_service(

name=request.name, project=request.project, allow_cache=False

)

assert_permissions(

resource=feature_service,

actions=[AuthzedAction.DELETE],

)

self.proxied_registry.delete_feature_service(

name=request.name, project=request.project, commit=request.commit

)

self._delete_openlineage_for_object(

request.name, request.project, "featureService"

)