Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def client(core_url, serving_url, allow_dirty):
# Get client for core and serving
client = Client(core_url=core_url, serving_url=serving_url)
client.create_project(PROJECT_NAME)
client.set_project(PROJECT_NAME)
# Ensure Feast core is active, but empty
if not allow_dirty:
feature_sets = client.list_feature_sets()
if len(feature_sets) > 0:
raise Exception("Feast cannot have existing feature sets registered. Exiting tests.")
return client
def client(core_url, serving_url, allow_dirty):
# Get client for core and serving
client = Client(core_url=core_url, serving_url=serving_url)
client.create_project(PROJECT_NAME)
client.set_project(PROJECT_NAME)
# Ensure Feast core is active, but empty
if not allow_dirty:
feature_sets = client.list_feature_sets()
if len(feature_sets) > 0:
raise Exception(
"Feast cannot have existing feature sets registered. Exiting tests."
)
return client
def version(client_only: bool, **kwargs):
"""
Displays version and connectivity information
"""
try:
feast_versions_dict = {
"sdk": {"version": str(pkg_resources.get_distribution("feast"))}
}
if not client_only:
feast_client = Client(
core_url=feast_config.get_config_property_or_fail(
"core_url", force_config=kwargs
),
serving_url=feast_config.get_config_property_or_fail(
"serving_url", force_config=kwargs
),
)
feast_versions_dict.update(feast_client.version())
print(json.dumps(feast_versions_dict))
except Exception as e:
_logger.error("Error initializing backend store")
_logger.exception(e)
sys.exit(1)
def feature_set_list():
"""
List all projects
"""
feast_client = Client(
core_url=feast_config.get_config_property_or_fail("core_url")
) # type: Client
table = []
for project in feast_client.list_projects():
table.append([project])
from tabulate import tabulate
print(tabulate(table, headers=["NAME"], tablefmt="plain"))
def feature_set_describe(name: str, version: int):
"""
Describe a feature set
"""
feast_client = Client(
core_url=feast_config.get_config_property_or_fail("core_url")
) # type: Client
fs = feast_client.get_feature_set(name=name, version=version)
if not fs:
print(
f'Feature set with name "{name}" and version "{version}" could not be found'
)
return
print(yaml.dump(yaml.safe_load(str(fs)), default_flow_style=False, sort_keys=False))
def feature_set_create(name):
"""
Create a feature set
"""
feast_client = Client(
core_url=feast_config.get_config_property_or_fail("core_url")
) # type: Client
feast_client.apply(FeatureSet(name=name))
def project_create(name: str):
"""
Create a project
"""
feast_client = Client(
core_url=feast_config.get_config_property_or_fail("core_url")
) # type: Client
feast_client.create_project(name)
def apply(filename):
"""
Apply a configuration to a resource by filename or stdin
"""
resources = [
ResourceFactory.get_resource(res_dict["kind"]).from_dict(res_dict)
for res_dict in yaml_loader(filename)
]
feast_client = Client(
core_url=feast_config.get_config_property_or_fail("core_url")
) # type: Client
feast_client.apply(resources)
def project_archive(name: str):
"""
Archive a project
"""
feast_client = Client(
core_url=feast_config.get_config_property_or_fail("core_url")
) # type: Client
feast_client.archive_project(name)
def feature_set_list():
"""
List all feature sets
"""
feast_client = Client(
core_url=feast_config.get_config_property_or_fail("core_url")
) # type: Client
table = []
for fs in feast_client.list_feature_sets():
table.append([fs.name, fs.version])
from tabulate import tabulate
print(tabulate(table, headers=["NAME", "VERSION"], tablefmt="plain"))