Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def dump(self, path):
"""Dump the import spec to the provided path
Arguments:
path (str): path to dump the spec to
"""
with open(path, "w") as f:
f.write(spec_to_yaml(self.spec))
print("Saved spec to {}".format(path))
def __str__(self):
"""Print the feature in yaml format
Returns:
str: yaml formatted representation of the entity
"""
return spec_to_yaml(self.__spec)
def __str__(self):
"""Print the feature in yaml format
Returns:
str: yaml formatted representation of the entity
"""
return spec_to_yaml(self.__spec)
def describe(self):
"""Print out the import spec.
"""
print(spec_to_yaml(self.spec))
"""
if apply_entity:
self._apply_entity(importer.entity)
if apply_features:
for feature in importer.features:
self._apply_feature(importer.features[feature])
if importer.require_staging:
print("Staging file to remote path {}".format(importer.remote_path))
importer.stage(feast_client=self)
request = JobServiceTypes.SubmitImportJobRequest(importSpec=importer.spec)
if name_override is not None:
request.name = name_override
print("Submitting job with spec:\n {}".format(spec_to_yaml(importer.spec)))
self._connect_core()
response = self._job_service_stub.SubmitJob(request)
print("Submitted job with id: {}".format(response.jobId))
return response.jobId
def __str__(self):
"""Return string representation of the feature group
Returns:
str: yaml formatted representation of the entity
"""
return spec_to_yaml(self.__spec)
def __str__(self):
"""Return string representation the storage in yaml format
Returns:
str: yaml formatted representation of the entity
"""
return spec_to_yaml(self.__spec)