Class RegistryServer

sdk/python/feast/registry_server.py:178–1345

class RegistryServer(RegistryServer_pb2_grpc.RegistryServerServicer):
    def __init__(self, registry: BaseRegistry) -> None:
        super().__init__()
        self.proxied_registry = registry

    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,
        )

        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