WIP: ConfigParser

.pre-commit-config.yaml

@@ -95,6 +95,7 @@ repos:
                 | util_text
                 | url_util
                 | version
+                | config_parser/parser
             ).py$
         additional_dependencies:
             - types-requests

MANIFEST.in

@@ -3,7 +3,7 @@ include *.rst
 include LICENSE.txt
 include NOTICE
 include pyproject.toml
-recursive-include src/snowflake/connector py.typed *.py *.pyx
+recursive-include src/snowflake/connector py.typed *.py *.pyx *.toml
 recursive-include src/snowflake/connector/vendored LICENSE*
 
 recursive-include src/snowflake/connector/cpp *.cpp *.hpp

setup.cfg

@@ -60,6 +60,8 @@ install_requires =
     certifi>=2017.4.17
     typing_extensions>=4.3,<5
     filelock>=3.5,<4
+    platformdirs
+    tomlkit
 include_package_data = True
 package_dir =
     =src

src/snowflake/connector/config_parser/__init__.py

@@ -0,0 +1,24 @@
+#
+# Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+#
+
+from __future__ import annotations
+
+from tomlkit import parse
+
+from ..constants import config_file
+from .parser import ConfigParser
+
+ROOT_PARSER = ConfigParser(
+    name="ROOT_PARSER",
+    file_path=config_file,
+)
+ROOT_PARSER.add_option(
+    name="connections",
+    _type=parse,
+)
+
+__all__ = [
+    "ConfigParser",
+    "ROOT_PARSER",
+]

src/snowflake/connector/config_parser/default_config.toml

@@ -0,0 +1,4 @@
+[connections.default]
+account = "accountname"
+user = "username"
+password = "password"

src/snowflake/connector/config_parser/parser.py

@@ -0,0 +1,180 @@
+#
+# Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+#
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterable
+from functools import wraps
+from pathlib import Path
+from typing import Callable, Literal, TypeVar
+
+import tomlkit
+from tomlkit import TOMLDocument
+from tomlkit.items import Table
+
+from ..errors import ConfigParserError, ConfigSourceError
+
+_T = TypeVar("_T")
+
+
+class ConfigOption:
+    def __init__(
+        self,
+        name: str,
+        _root_parser: ConfigParser,
+        _nest_path: list[str],
+        _type: Callable[[str], _T] | None = None,
+        choices: Iterable[_T] | None = None,
+        env_name: str | None | Literal[False] = None,
+    ) -> None:
+        """Create a config option that can read values from different locations.
+
+        Args:
+            name: The name of the ConfigOption
+            env_name: Environmental variable value should be read from, if not supplied, we'll construct this
+            type: A function that can turn str to the desired type, useful for reading value from environmental variable
+        """
+        self.name = name
+        self.type = _type
+        self.choices = choices
+        self._nest_path = _nest_path + [name]
+        self._root_parser: ConfigParser = _root_parser
+        self.env_name = env_name
+
+    def get(self):
+        """Retrieve a value of option."""
+        source = "environment variable"
+        value = self._get_env()
+        if value is None:
+            source = "configuration file"
+            value = self._get_config()
+        if self.choices and value not in self.choices:
+            raise ConfigSourceError(
+                f"The value of {self.generate_name()} read from "
+                f"{source} is not part of {self.choices}"
+            )
+        return value
+
+    def generate_name(self) -> str:
+        return ".".join(self._nest_path[1:])
+
+    def generate_env_name(self) -> str:
+        pieces = map(lambda e: e.upper(), self._nest_path[1:])
+        return f"SF{'_' + '_'.join(pieces)}"
+
+    def _get_env(self) -> str | _T | None:
+        if self.env_name is False:
+            return None
+        if self.env_name is not None:
+            env_name = self.env_name
+        else:
+            # Generate environment name if it wasn't not explicitly supplied,
+            #  and isn't disabled
+            env_name = self.generate_env_name()
+        if env_name not in os.environ:
+            return None
+        env_var = os.environ.get(env_name, None)
+        if env_var and self.type is not None:
+            return self.type(env_var)
+        return env_var
+
+    def _get_config(self):
+        e = self._root_parser._conf
+        for k in self._nest_path[1:]:
+            e = e[k]
+        if isinstance(e, Table):
+            # If we got a TOML table we probably want it in dictionary form
+            return e.value
+        return e
+
+
+class ConfigParser:
+    def __init__(
+        self,
+        *,
+        name: str,
+        file_path: Path | None = None,
+    ):
+        self.name = name
+        self.file_path = file_path
+        # Objects holding subparsers and options
+        self._options: dict[str, ConfigOption] = dict()
+        self._sub_parsers: dict[str, ConfigParser] = dict()
+        # Dictionary to cache read in config file
+        self._conf: TOMLDocument | None = None
+        # Information necessary to be able to nest elements
+        #  and add options in O(1)
+        self._root_parser: ConfigParser = self
+        self._nest_path = [name]
+
+    def read_config(
+        self,
+    ) -> None:
+        """Read and parse config file."""
+        if self.file_path is None:
+            raise ConfigParserError(
+                "ConfigParser is trying to read config file," " but it doesn't have one"
+            )
+        try:
+            self._conf = tomlkit.parse(self.file_path.read_text())
+        except Exception as e:
+            raise ConfigSourceError(
+                f'An unknown error happened while loading "{str(self.file_path)}'
+                f'", please see the error: {e}'
+            )
+
+    @wraps(ConfigOption.__init__)
+    def add_option(
+        self,
+        *args,
+        **kwargs,
+    ) -> None:
+        kwargs["_root_parser"] = self._root_parser
+        kwargs["_nest_path"] = self._nest_path
+        new_option = ConfigOption(
+            *args,
+            **kwargs,
+        )
+        self._check_child_conflict(new_option.name)
+        self._options[new_option.name] = new_option
+
+    def _check_child_conflict(self, name: str) -> None:
+        if name in (self._options.keys() | self._sub_parsers.keys()):
+            raise ConfigParserError(
+                f"'{name}' subparser, or option conflicts with a child element of '{self.name}'"
+            )
+
+    def add_subparser(self, other: ConfigParser) -> None:
+        self._check_child_conflict(other.name)
+        self._sub_parsers[other.name] = other
+
+        def _root_setter_helper(node: ConfigParser):
+            # Deal with ConfigParsers
+            node._root_parser = self._root_parser
+            node._nest_path = self._nest_path + node._nest_path
+            for sub_parser in node._sub_parsers.values():
+                _root_setter_helper(sub_parser)
+            # Deal with ConfigOptions
+            for option in node._options.values():
+                option._root_parser = self._root_parser
+                option._nest_path = self._nest_path + option._nest_path
+
+        _root_setter_helper(other)
+
+    def __getitem__(self, item: str) -> ConfigOption | ConfigParser:
+        if self._conf is None and (
+            self.file_path is not None
+            and self.file_path.exists()
+            and self.file_path.is_file()
+        ):
+            self.read_config()
+        if item in self._options:
+            return self._options[item].get()
+        if item not in self._sub_parsers:
+            raise ConfigSourceError(
+                "No ConfigParser, or ConfigOption can be found"
+                f" with the name '{item}'"
+            )
+        return self._sub_parsers[item]

src/snowflake/connector/constants.py

@@ -14,6 +14,21 @@
 if TYPE_CHECKING:
     from pyarrow import DataType
 
+from pathlib import Path
+from shutil import copyfile
+
+from platformdirs import PlatformDirs
+
+dirs = PlatformDirs(
+    appname="snowflake",
+    appauthor=False,
+    ensure_exists=True,
+)
+config_file = dirs.user_config_path / "config.toml"
+if not config_file.exists():
+    # Create default config file
+    default_config = Path(__file__).absolute().parent / "default_config.toml"
+    copyfile(default_config, config_file)
 
 DBAPI_TYPE_STRING = 0
 DBAPI_TYPE_BINARY = 1

src/snowflake/connector/errors.py

@@ -579,3 +579,17 @@ class PresignedUrlExpiredError(Error):
     """Exception for REST call to remote storage API failed because of expired presigned URL."""
 
     pass
+
+
+class ConfigSourceError(Error):
+    """Configuration source related errors.
+
+    Examples are environmental variable and configuration file.
+    """
+
+
+class ConfigParserError(Error):
+    """Configuration parser related errors.
+
+    These mean that ConfigParser is misused by a developer.
+    """