Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
project = self.project
else:
raise ValueError("No project has been configured.")
if version is None:
version = 0
try:
get_feature_set_response = self._core_service_stub.GetFeatureSet(
GetFeatureSetRequest(
project=project, name=name.strip(), version=int(version)
)
) # type: GetFeatureSetResponse
except grpc.RpcError as e:
raise grpc.RpcError(e.details())
return FeatureSet.from_proto(get_feature_set_response.feature_set)
f"your Feast Client."
)
else:
feature_set_proto.spec.project = self.project
# Convert the feature set to a request and send to Feast Core
try:
apply_fs_response = self._core_service_stub.ApplyFeatureSet(
ApplyFeatureSetRequest(feature_set=feature_set_proto),
timeout=GRPC_CONNECTION_TIMEOUT_APPLY,
) # type: ApplyFeatureSetResponse
except grpc.RpcError as e:
raise grpc.RpcError(e.details())
# Extract the returned feature set
applied_fs = FeatureSet.from_proto(apply_fs_response.feature_set)
# If the feature set has changed, update the local copy
if apply_fs_response.status == ApplyFeatureSetResponse.Status.CREATED:
print(
f'Feature set updated/created: "{applied_fs.name}:{applied_fs.version}"'
)
# If no change has been applied, do nothing
if apply_fs_response.status == ApplyFeatureSetResponse.Status.NO_CHANGE:
print(f"No change detected or applied: {feature_set.name}")
# Deep copy from the returned feature set to the local feature set
feature_set._update_from_feature_set(applied_fs)
if version is None:
version = "*"
filter = ListFeatureSetsRequest.Filter(
project=project, feature_set_name=name, feature_set_version=version
)
# Get latest feature sets from Feast Core
feature_set_protos = self._core_service_stub.ListFeatureSets(
ListFeatureSetsRequest(filter=filter)
) # type: ListFeatureSetsResponse
# Extract feature sets and return
feature_sets = []
for feature_set_proto in feature_set_protos.feature_sets:
feature_set = FeatureSet.from_proto(feature_set_proto)
feature_set._client = self
feature_sets.append(feature_set)
return feature_sets