Source code for ads.opctl.backend.base

#!/usr/bin/env python
# -*- coding: utf-8; -*-

# Copyright (c) 2022, 2023 Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v 1.0 as shown at

from abc import abstractmethod
from typing import Dict, Union

[docs] class UnsupportedRuntime(Exception): def __init__(self, runtime_type: str): super().__init__( f"The provided runtime: `{runtime_type}` " "is not supported by given resource." )
[docs] class Backend: """Interface for backend""" def __init__(self, config: Dict) -> None: self.config = config self.auth_type = config["execution"].get("auth", "api_key") self.profile = config["execution"].get("oci_profile", None) self.oci_config = config["execution"].get("oci_config", None)
[docs] @abstractmethod def run(self) -> Dict: """ Initiate a run. Returns ------- Dict """
[docs] def delete(self) -> None: """ Delete a remote run. Returns ------- None """
[docs] def watch(self) -> None: """ Stream logs from a remote run. Returns ------- None """
[docs] def cancel(self) -> None: """ Cancel a remote run. Returns ------- None """
[docs] def apply(self) -> Dict: """ Initiate Data Science service from YAML. Returns ------- Dict """
[docs] def activate(self) -> None: """ Activate a remote service. Returns ------- None """ raise NotImplementedError("`activate` has not been implemented yet.")
[docs] def deactivate(self) -> None: """ Deactivate a remote service. Returns ------- None """ raise NotImplementedError("`deactivate` has not been implemented yet.")
[docs] def run_diagnostics(self): """ Implement Diagnostics check appropriate for the backend """
[docs] def init( self, uri: Union[str, None] = None, overwrite: bool = False, **kwargs: Dict ) -> Union[str, None]: """Generates a YAML specification for the resource. Parameters ---------- overwrite: (bool, optional). Defaults to False. Overwrites the result specification YAML if exists. uri: (str, optional) The filename to save the resulting specification template YAML. **kwargs: Dict The optional arguments. runtime_type: str The resource runtime type. Returns ------- Union[str, None] The YAML specification for the given resource if `uri` was not provided. `None` otherwise. """ raise NotImplementedError( "The `init` has not been implemented yet for the given resource." )
[docs] def predict(self) -> None: """ Run model predict. Returns ------- None """ raise NotImplementedError("`predict` has not been implemented yet.")
[docs] class RuntimeFactory: """Base factory for runtime.""" _MAP = {}
[docs] @classmethod def get_runtime(cls, key: str, *args, **kwargs): if key not in cls._MAP: raise UnsupportedRuntime(key) return cls._MAP[key](*args, **kwargs)