feast-dev · mike0sv · Nov 16, 2020 · Nov 17, 2020 · Nov 20, 2020 · Nov 20, 2020
@@ -849,7 +849,9 @@ def ingest(
         try:
             if issubclass(type(feature_table.batch_source), FileSource):
                 file_url = feature_table.batch_source.file_options.file_url.rstrip("*")
-                _upload_to_file_source(file_url, with_partitions, dest_path)
+                _upload_to_file_source(
+                    file_url, with_partitions, dest_path, self._config
+                )
             if issubclass(type(feature_table.batch_source), BigQuerySource):
                 bq_table_ref = feature_table.batch_source.bigquery_options.table_ref
                 feature_table_timestamp_column = (
@@ -1004,6 +1006,7 @@ def get_historical_features(
                 entity_source = stage_entities_to_fs(
                     entity_source,
                     staging_location=self._config.get(opt.SPARK_STAGING_LOCATION),
+                    config=self._config,
                 )
 
         if self._use_job_service:

@@ -29,6 +29,7 @@
 from feast.constants import ConfigOptions as opt
 
 _logger = logging.getLogger(__name__)
+_UNSET = object()
 
 
 def _init_config(path: str):
@@ -50,17 +51,14 @@ def _init_config(path: str):
     os.makedirs(os.path.dirname(config_dir), exist_ok=True)
 
     # Create the configuration file itself
-    config = ConfigParser(defaults=opt().defaults())
+    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)
 
-    # Save the current configuration
-    config.write(open(path, "w"))
-
     return config
 
 
@@ -117,69 +115,66 @@ def __init__(
         self._config = config  # type: ConfigParser
         self._path = path  # type: str
 
-    def get(self, option):
+    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,
+        )
+
+    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._config.get(
-            CONFIG_FILE_SECTION,
-            option,
-            vars={**_get_feast_env_vars(), **self._options},
-        )
+        return self._get(option, default, self._config.get)
 
-    def getboolean(self, option):
+    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._config.getboolean(
-            CONFIG_FILE_SECTION,
-            option,
-            vars={**_get_feast_env_vars(), **self._options},
-        )
+        return self._get(option, default, self._config.getboolean)
 
-    def getint(self, option):
+    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._config.getint(
-            CONFIG_FILE_SECTION,
-            option,
-            vars={**_get_feast_env_vars(), **self._options},
-        )
+        return self._get(option, default, self._config.getint)
 
-    def getfloat(self, option):
+    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._config.getfloat(
-            CONFIG_FILE_SECTION,
-            option,
-            vars={**_get_feast_env_vars(), **self._options},
-        )
+        return self._get(option, default, self._config.getfloat)
 
     def set(self, option, value):
         """
@@ -211,7 +206,12 @@ def save(self):
         Save the current configuration to disk. This does not include
         environmental variables or initialized options
         """
-        self._config.write(open(self._path, "w"))
+        defaults = self._config.defaults()
+        try:
+            self._config._defaults = {}
+            self._config.write(open(self._path, "w"))
+        finally:
+            self._config._defaults = defaults
 
     def __str__(self):
         result = ""

@@ -125,6 +125,9 @@ class ConfigOptions(metaclass=ConfigMeta):
     #: Time to wait for historical feature requests before timing out.
     BATCH_FEATURE_REQUEST_WAIT_TIME_SECONDS: str = "600"
 
+    #: Endpoint URL for S3 storage_client
+    S3_ENDPOINT_URL: Optional[str] = None
+
     #: Authentication Provider - Google OpenID/OAuth
     #:
     #: Options: "google" / "oauth"

@@ -9,6 +9,7 @@
 import pyarrow as pa
 from pyarrow import parquet as pq
 
+from feast.config import Config
 from feast.staging.storage_client import get_staging_client
 
 
@@ -166,7 +167,7 @@ def _read_table_from_source(
 
 
 def _upload_to_file_source(
-    file_url: str, with_partitions: bool, dest_path: str
+    file_url: str, with_partitions: bool, dest_path: str, config: Config
 ) -> None:
     """
     Uploads data into a FileSource. Currently supports GCS, S3 and Local FS.
@@ -177,7 +178,7 @@ def _upload_to_file_source(
     from urllib.parse import urlparse
 
     uri = urlparse(file_url)
-    staging_client = get_staging_client(uri.scheme)
+    staging_client = get_staging_client(uri.scheme, config)
 
     if with_partitions:
         for path in glob.glob(os.path.join(dest_path, "**/*")):

@@ -7,6 +7,7 @@
 
 import pandas as pd
 
+from feast.config import Config
 from feast.data_format import ParquetFormat
 from feast.data_source import BigQuerySource, FileSource
 from feast.staging.storage_client import get_staging_client
@@ -18,15 +19,15 @@
 
 
 def stage_entities_to_fs(
-    entity_source: pd.DataFrame, staging_location: str
+    entity_source: pd.DataFrame, staging_location: str, config: Config
 ) -> FileSource:
     """
     Dumps given (entities) dataframe as parquet file and stage it to remote file storage (subdirectory of staging_location)
 
     :return: FileSource with remote destination path
     """
     entity_staging_uri = urlparse(os.path.join(staging_location, str(uuid.uuid4())))
-    staging_client = get_staging_client(entity_staging_uri.scheme)
+    staging_client = get_staging_client(entity_staging_uri.scheme, config)
     with tempfile.NamedTemporaryFile() as df_export_path:
         entity_source.to_parquet(df_export_path.name)
         bucket = (

@@ -24,6 +24,9 @@
 
 from google.auth.exceptions import DefaultCredentialsError
 
+from feast.config import Config
+from feast.constants import ConfigOptions as opt
+
 GS = "gs"
 S3 = "s3"
 LOCAL_FILE = "file"
@@ -144,15 +147,15 @@ class S3Client(AbstractStagingClient):
        Implementation of AbstractStagingClient for Aws S3 storage
     """
 
-    def __init__(self):
+    def __init__(self, endpoint_url: str = None):
         try:
             import boto3
         except ImportError:
             raise ImportError(
                 "Install package boto3 for s3 staging support"
                 "run ```pip install boto3```"
             )
-        self.s3_client = boto3.client("s3")
+        self.s3_client = boto3.client("s3", endpoint_url=endpoint_url)
 
     def download_file(self, uri: ParseResult) -> IO[bytes]:
         """
@@ -275,21 +278,38 @@ def upload_file(self, local_path: str, bucket: str, remote_path: str):
         shutil.copy(local_path, dest_fpath)
 
 
-storage_clients = {GS: GCSClient, S3: S3Client, LOCAL_FILE: LocalFSClient}
+def _s3_client(config: Config = None):
+    if config is None:
+        endpoint_url = None
+    else:
+        endpoint_url = config.get(opt.S3_ENDPOINT_URL, None)
+    return S3Client(endpoint_url=endpoint_url)
+
+
+def _gcs_client(config: Config = None):
+    return GCSClient()
+
+
+def _local_fs_client(config: Config = None):
+    return LocalFSClient()
+
+
+storage_clients = {GS: _gcs_client, S3: _s3_client, LOCAL_FILE: _local_fs_client}
 
 
-def get_staging_client(scheme):
+def get_staging_client(scheme, config: Config = None):
     """
     Initialization of a specific client object(GCSClient, S3Client etc.)
 
     Args:
         scheme (str): uri scheme: s3, gs or file
+        config (Config): additional configuration
 
     Returns:
         An object of concrete implementation of AbstractStagingClient
     """
     try:
-        return storage_clients[scheme]()
+        return storage_clients[scheme](config)
     except ValueError:
         raise Exception(
             f"Could not identify file scheme {scheme}. Only gs://, file:// and s3:// are supported"

@@ -103,6 +103,17 @@ def test_default_options(self):
         config = Config(path=path)
         assert config.get("CORE_URL") == "localhost:6565"
 
+    def test_defaults_are_not_written(self):
+        """
+        default values are not written to config file
+        """
+        fd, path = mkstemp()
+        config = Config(path=path)
+        config.set("option", "value")
+        config.save()
+        with open(path) as f:
+            assert f.read() == "[general]\noption = value\n\n"
+
     def test_type_casting(self):
         """
         Test type casting of strings to other types
@@ -117,6 +128,16 @@ def test_type_casting(self):
         assert config.getfloat("FLOAT_VAR") == 1.0
         assert config.getboolean("BOOLEAN_VAR") is True
 
+    def test_type_casting_of_defaults(self):
+        """
+        default values are casted as expected
+        """
+        fd, path = mkstemp()
+        config = Config(path=path)
+        assert isinstance(config.getboolean("enable_auth"), bool)
+        assert isinstance(config.getint("DATAPROC_EXECUTOR_INSTANCES"), int)
+        assert isinstance(config.getfloat("DATAPROC_EXECUTOR_INSTANCES"), float)
+
     def test_set_value(self):
         """
         Test type casting of strings to other types