Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __eq__(self, other):
return (
self.name == other.name
and self.type == other.type
and self.version == other.version
and self.description == other.description
and self.dependencies == other.dependencies
and self.methods == other.methods
and self.evaluator == other.evaluator
and self.schema_version == other.schema_version
and self.schema == other.schema
)
class ModelEndpoint(Endpoint):
"""Represents a model endpoint.
src_path : str
The local file path to the source of this object.
required_files : str
The local file path to the directory containing the
required files.
required_packages : str
The local file path to the directory containing the
required packages.
raise RuntimeError(
f"An endpoint with that name ({name}) already"
' exists. Use "override = True" to force update '
"an existing endpoint."
)
version = endpoint.version + 1
else:
version = 1
obj = self._gen_endpoint(name, obj, description, version, schema)
self._upload_endpoint(obj)
if version == 1:
self._service.add_endpoint(Endpoint(**obj))
else:
self._service.set_endpoint(Endpoint(**obj))
self._wait_for_endpoint_deployment(obj["name"], obj["version"])
' exists. Use "override = True" to force update '
"an existing endpoint."
)
version = endpoint.version + 1
else:
version = 1
obj = self._gen_endpoint(name, obj, description, version, schema)
self._upload_endpoint(obj)
if version == 1:
self._service.add_endpoint(Endpoint(**obj))
else:
self._service.set_endpoint(Endpoint(**obj))
self._wait_for_endpoint_deployment(obj["name"], obj["version"])
required_packages = RESTProperty(list)
required_packages_dst_path = RESTProperty(str)
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.type = "model"
def __eq__(self, other):
return (
super().__eq__(other)
and self.required_files == other.required_files
and self.required_packages == other.required_packages
)
class AliasEndpoint(Endpoint):
"""Represents an alias Endpoint.
target : str
The endpoint that this is an alias for.
"""
target = RESTProperty(str)
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.type = "alias"
class RESTServiceClient:
def get_endpoints(self, type=None):
"""Returns endpoints from the management API.
Parameters
----------
type : str
The type of endpoint to return. None will include all endpoints.
Other options are 'model' and 'alias'.
"""
result = {}
for name, attrs in self.service_client.GET("endpoints", {"type": type}).items():
endpoint = Endpoint.from_json(attrs)
endpoint.name = name
result[name] = endpoint
return result