#
# Copyright 2019 The Feast Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
import os
from configparser import ConfigParser, NoOptionError
from os.path import expanduser, join
from typing import Dict, Optional
from feast.constants import (
CONFIG_FEAST_ENV_VAR_PREFIX,
CONFIG_FILE_DEFAULT_DIRECTORY,
CONFIG_FILE_NAME,
CONFIG_FILE_SECTION,
FEAST_CONFIG_FILE_ENV,
)
from feast.constants import ConfigOptions as opt
_logger = logging.getLogger(__name__)
_UNSET = object()
def _init_config(path: str):
"""
Returns a ConfigParser that reads in a feast configuration file. If the
file does not exist it will be created.
Args:
path: Optional path to initialize as Feast configuration
Returns: ConfigParser of the Feast configuration file, with defaults
preloaded
"""
# Create the configuration file directory if needed
config_dir = os.path.dirname(path)
config_dir = config_dir.rstrip("/") + "/"
os.makedirs(os.path.dirname(config_dir), exist_ok=True)
# Create the configuration file itself
config = ConfigParser(defaults=opt().defaults(), allow_no_value=True)
if os.path.exists(path):
config.read(path)
# Store all configuration in a single section
if not config.has_section(CONFIG_FILE_SECTION):
config.add_section(CONFIG_FILE_SECTION)
return config
def _get_feast_env_vars():
"""
Get environmental variables that start with "FEAST_"
Returns: Dict of Feast environmental variables (stripped of prefix)
"""
feast_env_vars = {}
for key in os.environ.keys():
if key.upper().startswith(CONFIG_FEAST_ENV_VAR_PREFIX):
feast_env_vars[key[len(CONFIG_FEAST_ENV_VAR_PREFIX) :]] = os.environ[key]
return feast_env_vars
[docs]class Config:
"""
Maintains and provides access to Feast configuration
Configuration is stored as key/value pairs. The user can specify options
through either input arguments to this class, environmental variables, or
by setting the config in a configuration file
"""
def __init__(
self, options: Optional[Dict[str, str]] = None, path: Optional[str] = None,
):
"""
Configuration options are returned as follows (higher replaces lower)
1. Initialized options ("options" argument)
2. Environmental variables (reloaded on every "get")
3. Configuration file options (loaded once)
4. Default options (loaded once from memory)
Args:
options: (optional) A list of initialized/hardcoded options.
path: (optional) File path to configuration file
"""
if not path:
path = join(
expanduser("~"),
os.environ.get(FEAST_CONFIG_FILE_ENV, CONFIG_FILE_DEFAULT_DIRECTORY,),
CONFIG_FILE_NAME,
)
config = _init_config(path)
self._options = {}
if options and isinstance(options, dict):
self._options = options
self._config = config # type: ConfigParser
self._path = path # type: str
def _get(self, option, default, get_method):
fallback = {} if default is _UNSET else {"fallback": default}
return get_method(
CONFIG_FILE_SECTION,
option,
vars={**_get_feast_env_vars(), **self._options},
**fallback,
)
[docs] def get(self, option, default=_UNSET):
"""
Returns a single configuration option as a string
Args:
option: Name of the option
default: Default value to return if option is not found
Returns: String option that is returned
"""
return self._get(option, default, self._config.get)
[docs] def getboolean(self, option, default=_UNSET):
"""
Returns a single configuration option as a boolean
Args:
option: Name of the option
default: Default value to return if option is not found
Returns: Boolean option value that is returned
"""
return self._get(option, default, self._config.getboolean)
[docs] def getint(self, option, default=_UNSET):
"""
Returns a single configuration option as an integer
Args:
option: Name of the option
default: Default value to return if option is not found
Returns: Integer option value that is returned
"""
return self._get(option, default, self._config.getint)
[docs] def getfloat(self, option, default=_UNSET):
"""
Returns a single configuration option as an integer
Args:
option: Name of the option
default: Default value to return if option is not found
Returns: Float option value that is returned
"""
return self._get(option, default, self._config.getfloat)
[docs] def set(self, option, value):
"""
Sets a configuration option. Must be serializable to string
Args:
option: Option name to use as key
value: Value to store under option
"""
self._config.set(CONFIG_FILE_SECTION, option, value=str(value))
[docs] def exists(self, option):
"""
Tests whether a specific option is available
Args:
option: Name of the option to check
Returns: Boolean true/false whether the option is set
"""
try:
self.get(option=option)
return True
except NoOptionError:
return False
[docs] def save(self):
"""
Save the current configuration to disk. This does not include
environmental variables or initialized options
"""
defaults = self._config.defaults()
try:
self._config._defaults = {}
self._config.write(open(self._path, "w"))
finally:
self._config._defaults = defaults
def __str__(self):
result = ""
for section_name in self._config.sections():
result += "\n[" + section_name + "]\n"
for name, value in self._config.items(section_name):
result += name + " = " + value + "\n"
return result