Source code for datawork.api.config

"""Basic Option and Config functionality."""

import hashlib
import json
from abc import ABC


from .cache import Hashable
from .graph import Node


class Option(ABC):
    """An Option is a JSON serializable argument to a function."""

    value_type = None

    def __init__(self, desc=None, name=None, required=False, default=None):
        """Construct an 'Option' with name, default value, and description."""
        self.name = name
        self.prefix = ""
        self.desc = desc
        if default is not None and not isinstance(default, self.value_type):
            raise ValueError(f"Default value {default} is not of type {self.value_type} or None")
        self.default = default
        self._required = required
        # tracks whether value is set yet. Optional values are always set
        self._missing = required
        self._value = default

    def add_argument(self, parser):
        """Add an argument to an argparse 'ArgumentParser'."""
        parser.add_argument(
            f"--{self.prefix}{self.name}",
            default=self.default,
            required=self._required,
            type=self.value_type,
            help=self.desc,
        )

    def get_value(self):
        """Get the set value or raise a ValueError."""
        if self._missing:
            raise ValueError(
                "Required configuration option {self.prefix}{self.name} not set"
            )
        return self._value

    def set_value(self, value):
        """Implement a guarded setter for this Option type."""
        if not isinstance(value, self.value_type):
            raise TypeError(
                f"Cannot set option with type {str(self.value_type)} with "
                f"value of type {str(type(value))}"
            )
        self._value = value
        self._missing = False

    value = property(get_value, set_value)

    def __str__(self):
        """Show value and name for this Option."""
        return (f"{self.__class__.__name__}(name={self.name}, "
                f"_value={self._value})")


class Config(Hashable, Node):
    """A 'Config' is a collection of 'Option's."""

    def __init__(self, options):
        """Construct a Config with given list of 'Option's."""
        self._options = dict([(o.name, o) for o in options])
        for opt in options:
            if opt.name in self.__dict__:
                raise ValueError(f"Option {opt.name} is already present in Config")
            self.__dict__[opt.name] = opt

    def __setattr__(self, name, val):
        """Overload setattr to only allow setting of listed options."""
        if name == "_options":
            object.__setattr__(self, name, val)
        if name in self._options:
            self._options[name].value = val

    def __str__(self):
        """Convert to string by listing all options."""
        return f"{self.__class__.__name__}: {{" + \
               ", ".join([n + "=" + str(o)
                          for n, o in self._options.items()]) + "}"

    def to_dict(self):
        """Convert to dictionary to prepare for JSON conversion."""
        return {n: o.value for n, o in self._options.items()}

    def get_hash(self):
        """Return hash of the dictionary representation of this 'Config'."""
        return hashlib.md5("conf" +
                           json.dumps(self.to_dict, sort_keys=True)).hexdigest()

    def parents(self):
        """Return empty list of parents."""
        return []


class Configurable(Node):
    """A Configurable class contains a 'Config' attribute called '.config'."""

    OPTIONS = []

    def __init__(self):
        """Set a default .config using the class attribute '.OPTIONS'."""
        self.config = Config(self.OPTIONS)

    def parents(self):
        """Return `.config` as the only parent."""
        return [self.config]