Source code for ads.jobs.builders.base

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

# Copyright (c) 2021, 2023 Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
import copy
from typing import Any, Dict, TypeVar, Type
from ads.jobs.serializer import Serializable


Self = TypeVar("Self", bound="Builder")
"""Special type to represent the current enclosed class.

This type is used by factory class method or when a method returns ``self``.
"""


[docs] class Builder(Serializable): attribute_map = {} def __init__(self, spec: Dict = None, **kwargs) -> None: """To initialize the object, user can either pass in the specification as a dictionary or through keyword arguments. Parameters ---------- spec : dict, optional Object specification, by default None kwargs: dict Specification as keyword arguments. If spec contains the same key as the one in kwargs, the value from kwargs will be used. """ super().__init__() self._spec = self._load_default_properties() self._spec.update(self._standardize_spec(spec)) self._spec.update(self._standardize_spec(kwargs)) def _load_default_properties(self): """ Load default properties from environment variables, notebook session, etc. Should be implemented in the child classes. Returns ------- Dict A dictionary of default properties. """ return {} def _standardize_spec(self, spec) -> dict: if not spec: return {} snake_to_camel_map = {v: k for k, v in self.attribute_map.items()} for key in list(spec.keys()): if key not in self.attribute_map and key in snake_to_camel_map: spec[snake_to_camel_map[key]] = spec.pop(key) return spec
[docs] def set_spec(self: Self, k: str, v: Any) -> Self: """Sets a specification property for the object. Parameters ---------- k: str key, the name of the property. v: Any value, the value of the property. Returns ------- Self This method returns self to support chaining methods. """ if v is not None: self._spec[k] = v else: self._spec.pop(k, None) return self
[docs] def get_spec(self, key: str, default: Any = None) -> Any: """Gets the value of a specification property Parameters ---------- key : str The name of the property. default : Any, optional The default value to be used, if the property does not exist, by default None. Returns ------- Any The value of the property. """ return self._spec.get(key, default)
@property def kind(self) -> str: """The kind of the object as showing in YAML. Subclass should overwrite this value. """ return "builder" @property def type(self) -> str: """The type of the object as showing in YAML. This implementation returns the class name with the first letter coverted to lower case. """ class_name = self.__class__.__name__ return class_name[0].lower() + class_name[1:] if len(class_name) > 1 else ""
[docs] def to_dict(self, **kwargs) -> dict: """Converts the object to dictionary with kind, type and spec as keys. Parameters ---------- **kwargs: Dict The additional arguments. - filter_by_attribute_map: bool If True, then in the result will be included only the fields presented in the `attribute_map`. """ filter_by_attribute_map = kwargs.pop("filter_by_attribute_map", False) spec = { key: value for key, value in copy.deepcopy(self._spec).items() if not filter_by_attribute_map or key in self.attribute_map } for key, value in spec.items(): if hasattr(value, "to_dict"): value = value.to_dict() spec[key] = value return { "kind": self.kind, "type": self.type, "spec": spec, }
[docs] @classmethod def from_dict(cls: Type[Self], obj_dict: dict) -> Self: """Initialize the object from a Python dictionary""" return cls(spec=obj_dict.get("spec"))
def __repr__(self) -> str: """Displays the object as YAML.""" return self.to_yaml()
[docs] def build(self) -> Self: """Load default values from the environment for the job infrastructure. Should be implemented on the child level. """ return self