From 32778bb69615a67c2465f975a9361936999de232 Mon Sep 17 00:00:00 2001 From: Tiago Requeijo Date: Mon, 8 Jan 2024 18:50:27 -0500 Subject: [PATCH] Switch to ruff lint checker --- .github/workflows/run_tests.yml | 4 ++ config/__init__.py | 107 ++++++++++++++++---------------- config/configuration.py | 90 ++++++++++++--------------- config/configuration_set.py | 38 ++++++------ config/contrib/aws.py | 20 +++--- config/contrib/azure.py | 19 +++--- config/contrib/gcp.py | 23 +++---- config/contrib/vault.py | 19 +++--- config/helpers.py | 37 +++++------ pyproject.toml | 32 +++------- 10 files changed, 180 insertions(+), 209 deletions(-) diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index 1ceb591..a0e3ba7 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -6,6 +6,10 @@ on: - main pull_request: +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + env: POETRY_VERSION: 1.5.1 diff --git a/config/__init__.py b/config/__init__.py index 9862c89..9d55c80 100644 --- a/config/__init__.py +++ b/config/__init__.py @@ -31,8 +31,7 @@ def config( interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ) -> ConfigurationSet: - """ - Create a :class:`ConfigurationSet` instance from an iterable of configs. + """Create a :class:`ConfigurationSet` instance from an iterable of configs. :param configs: iterable of configurations :param prefix: prefix to filter environment variables with @@ -88,7 +87,7 @@ def config( if not isinstance(config_, (tuple, list)) or len(config_) == 0: raise ValueError( "configuration parameters must be a list of dictionaries," - " strings, or non-empty tuples/lists" + " strings, or non-empty tuples/lists", ) type_ = config_[0] if type_ == "dict": @@ -102,8 +101,10 @@ def config( params = list(config_[1:]) + default_args[(len(config_) - 2) :] instances.append( config_from_python( - *params, **default_kwargs, ignore_missing_paths=ignore_missing_paths - ) + *params, + **default_kwargs, + ignore_missing_paths=ignore_missing_paths, + ), ) elif type_ == "json": instances.append( @@ -111,7 +112,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, - ) + ), ) elif yaml and type_ == "yaml": instances.append( @@ -119,7 +120,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, - ) + ), ) elif toml and type_ == "toml": instances.append( @@ -127,7 +128,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, - ) + ), ) elif type_ == "ini": instances.append( @@ -135,7 +136,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, - ) + ), ) elif type_ == "dotenv": instances.append( @@ -143,7 +144,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, - ) + ), ) elif type_ == "path": instances.append( @@ -151,13 +152,15 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, - ) + ), ) else: raise ValueError(f'Unknown configuration type "{type_}"') return ConfigurationSet( - *instances, interpolate=interpolate, interpolate_type=interpolate_type + *instances, + interpolate=interpolate, + interpolate_type=interpolate_type, ) @@ -173,8 +176,7 @@ def __init__( interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ): - """ - Constructor. + """Class Constructor. :param prefix: prefix to filter environment variables with :param separator: separator to replace by dots @@ -215,8 +217,7 @@ def config_from_env( interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ) -> Configuration: - """ - Create a :class:`EnvConfiguration` instance from environment variables. + """Create a :class:`EnvConfiguration` instance from environment variables. :param prefix: prefix to filter environment variables with :param separator: separator to replace by dots @@ -246,8 +247,7 @@ def __init__( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ): - """ - Constructor. + """Class Constructor. :param path: path to read from :param remove_level: how many levels to remove from the resulting config @@ -278,7 +278,7 @@ def reload(self) -> None: ".".join( (x[0].split("/") + [y])[ (dotted_path_levels + self._remove_level) : - ] + ], ), ) for x in os.walk(path) @@ -288,7 +288,8 @@ def reload(self) -> None: result = {} for filename, key in files_keys: - result[key] = open(filename).read() + with open(filename) as f: + result[key] = f.read() except FileNotFoundError: if self._ignore_missing_paths: result = {} @@ -311,8 +312,7 @@ def config_from_path( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ) -> Configuration: - """ - Create a :class:`Configuration` instance from filesystem path. + """Create a :class:`Configuration` instance from filesystem path. :param path: path to read from :param remove_level: how many levels to remove from the resulting config @@ -343,8 +343,7 @@ def __init__( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ): - """ - Constructor. + """Class Constructor. :param data: path to a config file, or its contents :param read_from_file: whether to read from a file path or to interpret @@ -397,7 +396,8 @@ def _reload( """Reload the JSON data.""" if read_from_file: if isinstance(data, str): - result = json.load(open(data, "rt")) + with open(data, "rt") as f: + result = json.load(f) else: result = json.load(data) else: @@ -414,8 +414,7 @@ def config_from_json( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ) -> Configuration: - """ - Create a :class:`Configuration` instance from a JSON file. + """Create a :class:`Configuration` instance from a JSON file. :param data: path to a JSON file or contents :param read_from_file: whether to read from a file path or to interpret @@ -449,6 +448,7 @@ def __init__( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ): + """Class Constructor.""" self._section_prefix = section_prefix super().__init__( data=data, @@ -471,7 +471,8 @@ def optionxform(self, optionstr: str) -> str: if read_from_file: if isinstance(data, str): - data = open(data, "rt").read() + with open(data, "rt") as f: + data = f.read() else: data = data.read() data = cast(str, data) @@ -496,8 +497,7 @@ def config_from_ini( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ) -> Configuration: - """ - Create a :class:`Configuration` instance from an INI file. + """Create a :class:`Configuration` instance from an INI file. :param data: path to an INI file or contents :param read_from_file: whether to read from a file path or to interpret @@ -533,6 +533,7 @@ def __init__( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ): + """Class Constructor.""" self._prefix = prefix self._separator = separator super().__init__( @@ -552,7 +553,8 @@ def _reload( """Reload the .env data.""" if read_from_file: if isinstance(data, str): - data = open(data, "rt").read() + with open(data, "rt") as f: + data = f.read() else: data = data.read() data = cast(str, data) @@ -568,8 +570,6 @@ def _reload( if k.startswith(self._prefix) } - print(self._prefix, self._separator, result) - self._config = self._flatten_dict(result) @@ -584,8 +584,7 @@ def config_from_dotenv( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ) -> Configuration: - """ - Create a :class:`Configuration` instance from a .env type file. + """Create a :class:`Configuration` instance from a .env type file. :param data: path to a .env type file or contents :param read_from_file: whether to read from a file path or to interpret @@ -621,8 +620,7 @@ def __init__( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ): - """ - Constructor. + """Class Constructor. :param module: a module or path string :param prefix: prefix to use to filter object names @@ -696,8 +694,7 @@ def config_from_python( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ) -> Configuration: - """ - Create a :class:`Configuration` instance from the objects in a Python module. + """Create a :class:`Configuration` instance from the objects in a Python module. :param module: a module or path string :param prefix: prefix to use to filter object names @@ -724,8 +721,7 @@ def config_from_dict( interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ) -> Configuration: - """ - Create a :class:`Configuration` instance from a dictionary. + """Create a :class:`Configuration` instance from a dictionary. :param data: dictionary with string keys :param lowercase_keys: whether to convert every key to lower case. @@ -741,10 +737,11 @@ def config_from_dict( def create_path_from_config( - path: str, cfg: Configuration, remove_level: int = 1 + path: str, + cfg: Configuration, + remove_level: int = 1, ) -> Configuration: - """ - Output a path configuration from a :class:`Configuration` instance. + """Output a path configuration from a :class:`Configuration` instance. :param path: path to create the config files in :param cfg: :class:`Configuration` instance @@ -776,9 +773,10 @@ def __init__( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ): - if yaml is None: + """Class Constructor.""" + if yaml is None: # pragma: no cover raise ImportError( - "Dependency is not found, but required by this class." + "Dependency is not found, but required by this class.", ) super().__init__( data=data, @@ -792,7 +790,8 @@ def __init__( def _reload(self, data: Union[str, TextIO], read_from_file: bool = False) -> None: """Reload the YAML data.""" if read_from_file and isinstance(data, str): - loaded = yaml.load(open(data, "rt"), Loader=yaml.FullLoader) + with open(data, "rt") as f: + loaded = yaml.load(f, Loader=yaml.FullLoader) else: loaded = yaml.load(data, Loader=yaml.FullLoader) if not isinstance(loaded, Mapping): @@ -809,8 +808,7 @@ def config_from_yaml( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ) -> Configuration: - """ - Return a Configuration instance from YAML files. + """Return a Configuration instance from YAML files. :param data: string or file :param read_from_file: whether `data` is a file or a YAML formatted string @@ -843,9 +841,10 @@ def __init__( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ): - if toml is None: + """Class Constructor.""" + if toml is None: # pragma: no cover raise ImportError( - "Dependency is not found, but required by this class." + "Dependency is not found, but required by this class.", ) self._section_prefix = section_prefix @@ -862,7 +861,8 @@ def _reload(self, data: Union[str, TextIO], read_from_file: bool = False) -> Non """Reload the TOML data.""" if read_from_file: if isinstance(data, str): - loaded = toml.load(open(data, "rt")) + with open(data, "rt") as f: + loaded = toml.load(f) else: loaded = toml.load(data) else: @@ -889,8 +889,7 @@ def config_from_toml( interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ignore_missing_paths: bool = False, ) -> Configuration: - """ - Return a Configuration instance from TOML files. + """Return a Configuration instance from TOML files. :param data: string or file :param read_from_file: whether `data` is a file or a TOML formatted string diff --git a/config/configuration.py b/config/configuration.py index e682ce1..33f0906 100644 --- a/config/configuration.py +++ b/config/configuration.py @@ -33,8 +33,7 @@ class Configuration: - """ - Configuration class. + """Configuration class. The Configuration class takes a dictionary input with keys such as @@ -55,8 +54,7 @@ def __init__( interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ): - """ - Constructor. + """Class Constructor. :param config_: a mapping of configuration values. Keys need to be strings. :param lowercase_keys: whether to convert every key to lower case. @@ -74,8 +72,7 @@ def __eq__(self, other): # type: ignore return self.as_dict() == Configuration(other).as_dict() def _filter_dict(self, d: Dict[str, Any], prefix: str) -> Dict[str, Any]: - """ - Filter a dictionary and return the items that are prefixed by :attr:`prefix`. + """Filter a dictionary and return the items that are prefixed by :attr:`prefix`. :param d: dictionary :param prefix: prefix to filter on @@ -95,8 +92,7 @@ def _filter_dict(self, d: Dict[str, Any], prefix: str) -> Dict[str, Any]: } def _flatten_dict(self, d: Mapping[str, Any]) -> Dict[str, Any]: - """ - Flatten one level of a dictionary. + """Flatten one level of a dictionary. :param d: dict :return: a flattened dict @@ -127,12 +123,11 @@ def _flatten_dict(self, d: Mapping[str, Any]) -> Dict[str, Any]: return result def _get_subset(self, prefix: str) -> Union[Dict[str, Any], Any]: - """ - Return the subset of the config dictionary whose keys start with :attr:`prefix`. + """Return the subset of the config dictionary whose keys start with :attr:`prefix`. :param prefix: string :return: dict - """ + """ # noqa: E501 d = { k[(len(prefix) + 1) :]: v for k, v in self._config.items() @@ -170,11 +165,10 @@ def __getattr__(self, item: str) -> Any: # noqa: D105 try: return self[item] except KeyError: - raise AttributeError(item) + raise AttributeError(item) from None def get(self, key: str, default: Any = None) -> Union[dict, Any]: - """ - Get the configuration values corresponding to :attr:`key`. + """Get the configuration values corresponding to :attr:`key`. :param key: key to retrieve :param default: default value in case the key is missing @@ -192,20 +186,18 @@ def as_attrdict(self) -> AttributeDict: { x: Configuration(v).as_attrdict() if isinstance(v, Mapping) else v for x, v in self.items(levels=1) - } + }, ) def get_bool(self, item: str) -> bool: - """ - Get the item value as a bool. + """Get the item value as a bool. :param item: key """ return as_bool(self[item]) def get_str(self, item: str, fmt: str = "{}") -> str: - """ - Get the item value as an int. + """Get the item value as an int. :param item: key :param fmt: format to use @@ -213,40 +205,35 @@ def get_str(self, item: str, fmt: str = "{}") -> str: return fmt.format(self[item]) def get_int(self, item: str) -> int: - """ - Get the item value as an int. + """Get the item value as an int. :param item: key """ return int(self[item]) def get_float(self, item: str) -> float: - """ - Get the item value as a float. + """Get the item value as a float. :param item: key """ return float(self[item]) def get_list(self, item: str) -> List[Any]: - """ - Get the item value as a list. + """Get the item value as a list. :param item: key """ return list(self[item]) def get_dict(self, item: str) -> dict: - """ - Get the item values as a dictionary. + """Get the item values as a dictionary. :param item: key """ return dict(self._get_subset(item)) def base64encode(self, item: str) -> bytes: - """ - Get the item value as a Base64 encoded bytes instance. + """Get the item value as a Base64 encoded bytes instance. :param item: key """ @@ -255,8 +242,7 @@ def base64encode(self, item: str) -> bytes: return base64.b64encode(b) def base64decode(self, item: str) -> bytes: - """ - Get the item value as a Base64 decoded bytes instance. + """Get the item value as a Base64 decoded bytes instance. :param item: key """ @@ -265,7 +251,8 @@ def base64decode(self, item: str) -> bytes: return base64.b64decode(b, validate=True) def keys( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, KeysView[str]]: """Return a set-like object providing a view on the configuration keys.""" assert levels is None or levels > 0 @@ -279,12 +266,13 @@ def keys( { ".".join(x.split(".")[:levels]) for x in set(self.as_dict().keys()) - } + }, ), ) def values( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ValuesView[Any]]: """Return a set-like object providing a view on the configuration values.""" assert levels is None or levels > 0 @@ -295,7 +283,8 @@ def values( return dict(self.items(levels=levels)).values() def items( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ItemsView[str, Any]]: """Return a set-like object providing a view on the configuration items.""" assert levels is None or levels > 0 @@ -311,7 +300,9 @@ def __iter__(self) -> Iterator[Tuple[str, Any]]: # noqa: D105 def __reversed__(self) -> Iterator[Tuple[str, Any]]: # noqa: D105 if version_info < (3, 8): - return OrderedDict(reversed(list(self.items()))) # type: ignore # pragma: no cover + return OrderedDict( + reversed(list(self.items())), + ) # type: ignore # pragma: no cover else: return reversed(dict(self.items())) # type: ignore @@ -322,8 +313,7 @@ def __setitem__(self, key: str, value: Any) -> None: # noqa: D105 self.update({key: value}) def __delitem__(self, prefix: str) -> None: # noqa: D105 - """ - Filter a dictionary and delete the items that are prefixed by :attr:`prefix`. + """Filter a dictionary and delete the items that are prefixed by :attr:`prefix`. :param prefix: prefix to filter on to delete keys """ @@ -353,8 +343,7 @@ def copy(self) -> "Configuration": return Configuration(self._config) def pop(self, prefix: str, value: Any = None) -> Any: - """ - Remove keys with the specified prefix and return the corresponding value. + """Remove keys with the specified prefix and return the corresponding value. If the prefix is not found a KeyError is raised. """ @@ -367,8 +356,7 @@ def pop(self, prefix: str, value: Any = None) -> Any: return value def setdefault(self, key: str, default: Any = None) -> Any: - """ - Insert key with a value of default if key is not in the Configuration. + """Insert key with a value of default if key is not in the Configuration. Return the value for key if key is in the Configuration, else default. """ @@ -383,8 +371,7 @@ def update(self, other: Mapping[str, Any]) -> None: self._config.update(self._flatten_dict(other)) def reload(self) -> None: # pragma: no cover - """ - Reload the configuration. + """Reload the configuration. This method is not implemented for simple Configuration objects and is intended only to be used in subclasses. @@ -392,12 +379,18 @@ def reload(self) -> None: # pragma: no cover raise NotImplementedError() def validate( - self, schema: Any, raise_on_error: bool = False, **kwargs: Mapping[str, Any] + self, + schema: Any, + raise_on_error: bool = False, + **kwargs: Mapping[str, Any], ) -> bool: + """Validate the current config using JSONSchema.""" try: - from jsonschema import validate, ValidationError + from jsonschema import ValidationError, validate except ImportError: # pragma: no cover - raise RuntimeError("Validation requires the `jsonschema` library.") + raise RuntimeError( + "Validation requires the `jsonschema` library.", + ) from None try: validate(self.as_dict(), schema, **kwargs) except ValidationError as err: @@ -408,8 +401,7 @@ def validate( @contextmanager def dotted_iter(self) -> Iterator["Configuration"]: - """ - Context manager for dotted iteration. + """Context manager for dotted iteration. This context manager changes all the iterator-related functions to include every nested (dotted) key instead of just the top level. diff --git a/config/configuration_set.py b/config/configuration_set.py index 46ab9c8..7d8115a 100644 --- a/config/configuration_set.py +++ b/config/configuration_set.py @@ -1,8 +1,8 @@ """ConfigurationSet class.""" +import contextlib from typing import ( Any, - Dict, ItemsView, Iterable, KeysView, @@ -11,7 +11,6 @@ Optional, Union, ValuesView, - cast, ) from .configuration import Configuration @@ -19,8 +18,7 @@ class ConfigurationSet(Configuration): - """ - Configuration Sets. + """Configuration Sets. A class that combines multiple :class:`Configuration` instances in a hierarchical manner. @@ -30,25 +28,26 @@ def __init__( self, *configs: Configuration, interpolate: InterpolateType = False, - interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD + interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, ): # noqa: D107 + """Class Constructor.""" self._interpolate = {} if interpolate is True else interpolate self._interpolate_type = interpolate_type try: self._configs: List[Configuration] = list(configs) except Exception: # pragma: no cover raise ValueError( - "configs should be a non-empty iterable of Configuration objects" - ) + "configs should be a non-empty iterable of Configuration objects", + ) from None if not self._configs: # pragma: no cover raise ValueError( - "configs should be a non-empty iterable of Configuration objects" + "configs should be a non-empty iterable of Configuration objects", ) if not all( isinstance(x, Configuration) for x in self._configs ): # pragma: no cover raise ValueError( - "configs should be a non-empty iterable of Configuration objects" + "configs should be a non-empty iterable of Configuration objects", ) self._writable = False self._default_levels = 1 @@ -113,8 +112,7 @@ def __getattr__(self, item: str) -> Union[Configuration, Any]: # noqa: D105 return self._from_configs("__getattr__", item) def get(self, key: str, default: Any = None) -> Union[dict, Any]: - """ - Get the configuration values corresponding to :attr:`key`. + """Get the configuration values corresponding to :attr:`key`. :param key: key to retrieve :param default: default value in case the key is missing @@ -133,15 +131,15 @@ def as_dict(self) -> dict: return result def get_dict(self, item: str) -> dict: - """ - Get the item values as a dictionary. + """Get the item values as a dictionary. :param item: key """ - return Configuration({k: v for k, v in dict(self[item]).items()}).as_dict() + return Configuration(dict(dict(self[item]).items())).as_dict() def keys( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, KeysView[str]]: """Return a set-like object providing a view on the configuration keys.""" if self._default_levels: @@ -150,7 +148,8 @@ def keys( return cfg.keys(levels) def values( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ValuesView[Any]]: """Return a set-like object providing a view on the configuration values.""" if self._default_levels: @@ -159,7 +158,8 @@ def values( return cfg.values(levels) def items( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ItemsView[str, Any]]: """Return a set-like object providing a view on the configuration items.""" if self._default_levels: @@ -202,10 +202,8 @@ def update(self, other: Mapping[str, Any]) -> None: def reload(self) -> None: """Reload the underlying configuration instances.""" for cfg in self._configs: - try: + with contextlib.suppress(NotImplementedError): cfg.reload() - except NotImplementedError: - pass def __repr__(self) -> str: # noqa: D105 return "" % hex(id(self)) diff --git a/config/contrib/aws.py b/config/contrib/aws.py index 1cc0388..effc301 100644 --- a/config/contrib/aws.py +++ b/config/contrib/aws.py @@ -5,10 +5,8 @@ from typing import Any, Dict, Optional import boto3 - from botocore.exceptions import ClientError - from .. import Configuration, InterpolateType @@ -21,8 +19,7 @@ def __init__(self, value: Dict[str, Any], ts: float): # noqa: D107 class AWSSecretsManagerConfiguration(Configuration): - """ - AWS Configuration class. + """AWS Configuration class. The AWS Configuration class takes AWS Secrets Manager credentials and behaves like a drop-in replacement for the regular Configuration class. @@ -40,8 +37,7 @@ def __init__( lowercase_keys: bool = False, interpolate: InterpolateType = False, ) -> None: - """ - Constructor. + """Class Constructor. :param secret_name: Name of the secret :param aws_access_key_id: AWS Access Key ID @@ -74,31 +70,31 @@ def _config(self) -> Dict[str, Any]: # type: ignore return self._secret.value try: get_secret_value_response = self._client.get_secret_value( - SecretId=self._secret_name + SecretId=self._secret_name, ) except ClientError as e: # pragma: no cover if e.response["Error"]["Code"] == "DecryptionFailureException": # Secrets Manager can't decrypt the protected secret text using # the provided KMS key. # Deal with the exception here, and/or rethrow at your discretion. - raise RuntimeError("Cannot read the AWS secret") + raise RuntimeError("Cannot read the AWS secret") from None elif e.response["Error"]["Code"] == "InternalServiceErrorException": # An error occurred on the server side. # Deal with the exception here, and/or rethrow at your discretion. - raise RuntimeError("Cannot read the AWS secret") + raise RuntimeError("Cannot read the AWS secret") from None elif e.response["Error"]["Code"] == "InvalidParameterException": # You provided an invalid value for a parameter. # Deal with the exception here, and/or rethrow at your discretion. - raise RuntimeError("Cannot read the AWS secret") + raise RuntimeError("Cannot read the AWS secret") from None elif e.response["Error"]["Code"] == "InvalidRequestException": # You provided a parameter value that is not valid for the current # state of the resource. # Deal with the exception here, and/or rethrow at your discretion. - raise RuntimeError("Cannot read the AWS secret") + raise RuntimeError("Cannot read the AWS secret") from None elif e.response["Error"]["Code"] == "ResourceNotFoundException": # We can't find the resource that you asked for. # Deal with the exception here, and/or rethrow at your discretion. - raise RuntimeError("Cannot read the AWS secret") + raise RuntimeError("Cannot read the AWS secret") from None else: # Decrypts secret using the associated KMS CMK. # Depending on whether the secret is a string or binary, one of these diff --git a/config/contrib/azure.py b/config/contrib/azure.py index a20d4b7..f6d32e7 100644 --- a/config/contrib/azure.py +++ b/config/contrib/azure.py @@ -7,7 +7,6 @@ from azure.identity import ClientSecretCredential from azure.keyvault.secrets import SecretClient - from .. import Configuration, InterpolateType @@ -20,8 +19,7 @@ def __init__(self, value: Optional[str], ts: float): # noqa: D107 class AzureKeyVaultConfiguration(Configuration): - """ - Azure Configuration class. + """Azure Configuration class. The Azure Configuration class takes Azure KeyVault credentials and behaves like a drop-in replacement for the regular Configuration class. @@ -42,8 +40,7 @@ def __init__( cache_expiration: int = 5 * 60, interpolate: InterpolateType = False, ) -> None: - """ - Constructor. + """Class Constructor. :param az_client_id: Client ID :param az_client_secret: Client Secret @@ -93,8 +90,7 @@ def __getattr__(self, item: str) -> Any: # noqa: D105 return secret def get(self, key: str, default: Any = None) -> Union[dict, Any]: - """ - Get the configuration values corresponding to :attr:`key`. + """Get the configuration values corresponding to :attr:`key`. :param key: key to retrieve :param default: default value in case the key is missing @@ -107,7 +103,8 @@ def get(self, key: str, default: Any = None) -> Union[dict, Any]: return secret def keys( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, KeysView[str]]: """Return a set-like object providing a view on the configuration keys.""" assert not levels # Azure Key Vaults don't support separators @@ -117,7 +114,8 @@ def keys( ) def values( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ValuesView[Any]]: """Return a set-like object providing a view on the configuration values.""" assert not levels # Azure Key Vaults don't support separators @@ -130,7 +128,8 @@ def values( ) def items( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ItemsView[str, Any]]: """Return a set-like object providing a view on the configuration items.""" assert not levels # Azure Key Vaults don't support separators diff --git a/config/contrib/gcp.py b/config/contrib/gcp.py index df69d20..35b00d6 100644 --- a/config/contrib/gcp.py +++ b/config/contrib/gcp.py @@ -20,8 +20,7 @@ def __init__(self, value: str, ts: float): # noqa: D107 class GCPSecretManagerConfiguration(Configuration): - """ - GCP Secret Manager Configuration class. + """GCP Secret Manager Configuration class. The GCP Secret Manager Configuration class takes GCP Secret Manager credentials and behaves like a drop-in replacement for the regular Configuration class. @@ -41,8 +40,7 @@ def __init__( cache_expiration: int = 5 * 60, interpolate: InterpolateType = False, ) -> None: - """ - Constructor. + """Class Constructor. See https://googleapis.dev/python/secretmanager/latest/gapic/v1/api.html#google.cloud.secretmanager_v1.SecretManagerServiceClient for more details on credentials and options. @@ -53,7 +51,8 @@ def __init__( :param cache_expiration: Cache expiration (in seconds) """ # noqa: E501 self._client = secretmanager_v1.SecretManagerServiceClient( - credentials=credentials, client_options=client_options + credentials=credentials, + client_options=client_options, ) self._project_id = project_id self._parent = f"projects/{project_id}" @@ -70,7 +69,7 @@ def _get_secret(self, key: str) -> Optional[str]: try: path = f"projects/{self._project_id}/secrets/{key}/versions/latest" secret = self._client.access_secret_version( - request={"name": path} + request={"name": path}, ).payload.data.decode() self._cache[key] = Cache(value=secret, ts=now) return secret @@ -94,8 +93,7 @@ def __getattr__(self, item: str) -> Any: # noqa: D105 return secret def get(self, key: str, default: Any = None) -> Union[dict, Any]: - """ - Get the configuration values corresponding to :attr:`key`. + """Get the configuration values corresponding to :attr:`key`. :param key: key to retrieve :param default: default value in case the key is missing @@ -108,7 +106,8 @@ def get(self, key: str, default: Any = None) -> Union[dict, Any]: return secret def keys( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, KeysView[str]]: """Return a set-like object providing a view on the configuration keys.""" assert not levels # GCP Secret Manager secrets don't support separators @@ -121,7 +120,8 @@ def keys( ) def values( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ValuesView[Any]]: """Return a set-like object providing a view on the configuration values.""" assert not levels # GCP Secret Manager secrets don't support separators @@ -134,7 +134,8 @@ def values( ) def items( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ItemsView[str, Any]]: """Return a set-like object providing a view on the configuration items.""" assert not levels # GCP Secret Manager secrets don't support separators diff --git a/config/contrib/vault.py b/config/contrib/vault.py index 0e9db11..8f44b49 100644 --- a/config/contrib/vault.py +++ b/config/contrib/vault.py @@ -16,7 +16,6 @@ import hvac from hvac.exceptions import InvalidPath - from .. import Configuration, InterpolateType, config_from_dict @@ -29,8 +28,7 @@ def __init__(self, value: Dict[str, Any], ts: float): # noqa: D107 class HashicorpVaultConfiguration(Configuration): - """ - Hashicorp Vault Configuration class. + """Hashicorp Vault Configuration class. The Hashicorp Vault Configuration class takes Vault credentials and behaves like a drop-in replacement for the regular Configuration class. @@ -48,8 +46,7 @@ def __init__( interpolate: InterpolateType = False, **kwargs: Mapping[str, Any], ) -> None: - """ - Constructor. + """Class Constructor. See https://developer.hashicorp.com/vault/docs/get-started/developer-qs. """ # noqa: E501 @@ -97,8 +94,7 @@ def __getattr__(self, item: str) -> Any: # noqa: D105 return Configuration(secret) def get(self, key: str, default: Any = None) -> Union[dict, Any]: - """ - Get the configuration values corresponding to :attr:`key`. + """Get the configuration values corresponding to :attr:`key`. :param key: key to retrieve :param default: default value in case the key is missing @@ -110,7 +106,8 @@ def get(self, key: str, default: Any = None) -> Union[dict, Any]: return default def keys( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, KeysView[str]]: """Return a set-like object providing a view on the configuration keys.""" assert not levels # Vault secrets don't support separators @@ -120,7 +117,8 @@ def keys( ) def values( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ValuesView[Any]]: """Return a set-like object providing a view on the configuration values.""" assert not levels # GCP Secret Manager secrets don't support separators @@ -133,7 +131,8 @@ def values( ) def items( - self, levels: Optional[int] = None + self, + levels: Optional[int] = None, ) -> Union["Configuration", Any, ItemsView[str, Any]]: """Return a set-like object providing a view on the configuration items.""" assert not levels # GCP Secret Manager secrets don't support separators diff --git a/config/helpers.py b/config/helpers.py index 26930ac..e0a9d0a 100644 --- a/config/helpers.py +++ b/config/helpers.py @@ -4,7 +4,6 @@ from enum import Enum from typing import Any, Dict, List, Set, Tuple, Union - TRUTH_TEXT = frozenset(("t", "true", "y", "yes", "on", "1")) FALSE_TEXT = frozenset(("f", "false", "n", "no", "off", "0", "")) PROTECTED_KEYS = frozenset(("secret", "password", "passwd", "pwd", "token")) @@ -33,15 +32,14 @@ def __getattr__(self, key: Any) -> Any: # noqa: D105 return self[key] except KeyError: # to conform with __getattr__ spec - raise AttributeError(key) + raise AttributeError(key) from None def __setattr__(self, key: Any, value: Any) -> None: # noqa: D105 self[key] = value def as_bool(s: Any) -> bool: - """ - Boolean value from an object. + """Boolean value from an object. Return the boolean value ``True`` if the case-lowered value of string input ``s`` is a `truthy string`. If ``s`` is already one of the @@ -58,8 +56,7 @@ def as_bool(s: Any) -> bool: def clean(key: str, value: Any, mask: str = "******") -> Any: - """ - Mask a value if needed. + """Mask a value if needed. :param key: key :param value: value to hide @@ -80,14 +77,13 @@ def clean(key: str, value: Any, mask: str = "******") -> Any: return value else: return url._replace( - netloc="{}:{}@{}".format(url.username, mask, url.hostname) + netloc="{}:{}@{}".format(url.username, mask, url.hostname), ).geturl() return value def interpolate_standard(text: str, d: dict, found: Set[Tuple[str, ...]]) -> str: - """ - Return the string interpolated as many times as needed. + """Return the string interpolated as many times as needed. :param text: string possibly containing an interpolation pattern :param d: dictionary @@ -97,7 +93,7 @@ def interpolate_standard(text: str, d: dict, found: Set[Tuple[str, ...]]) -> str return text variables = tuple( - sorted(x[1] for x in string.Formatter().parse(text) if x[1] is not None) + sorted(x[1] for x in string.Formatter().parse(text) if x[1] is not None), ) if not variables: @@ -120,8 +116,7 @@ def interpolate_deep( levels: Dict[str, int], method: InterpolateEnumType, ) -> str: - """ - Return the string interpolated as many times as needed. + """Return the string interpolated as many times as needed. :param attr: attribute name :param text: string possibly containing an interpolation pattern @@ -159,15 +154,19 @@ def interpolate_deep( else d ) resolved[variable] = interpolate_deep( - attr, d[level][variable], new_d, resolved, levels, method + attr, + d[level][variable], + new_d, + resolved, + levels, + method, ) return text.format(**resolved) def flatten(d: List[dict]) -> dict: - """ - Flatten a list of dictionaries. + """Flatten a list of dictionaries. :param d: dictionary list """ @@ -177,10 +176,12 @@ def flatten(d: List[dict]) -> dict: def interpolate_object( - attr: str, obj: Any, d: List[dict], method: InterpolateEnumType + attr: str, + obj: Any, + d: List[dict], + method: InterpolateEnumType, ) -> Any: - """ - Return the interpolated object. + """Return the interpolated object. :param attr: attribute name :param obj: object to interpolate diff --git a/pyproject.toml b/pyproject.toml index 021e7a7..48d61d7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,23 +23,18 @@ toml = { version = "^0.10.0", optional = true } jsonschema = { version = "^4.18.6", optional = true } [tool.poetry.group.dev.dependencies] -flake8-blind-except = "^0.2.0" -flake8-bugbear = "^23.7.10" -flake8-builtins = "^2.1.0" -flake8-comprehensions = "^3.3.1" -flake8-docstrings = "^1.3" -flake8-import-order = "^0.18.0" mypy = "^1.4.1" pydocstyle = "^6.0" pytest = "^7.4.0" pytest-black = "^0.3.12" pytest-cov = "^4.1.0" -pytest-flake8 = "^1.1.1" pytest-mypy = "^0.10.3" pytest-mock = "^3.5.0" sphinx = "^7.1.2" sphinx-autodoc-typehints = "^1.24.0" black = "^23.7.0" +ruff = "^0.0.284" +pytest-ruff = "^0.2.1" [tool.poetry.extras] aws = ["boto3"] @@ -50,8 +45,11 @@ vault = ["hvac"] yaml = ["pyyaml"] validation = ["jsonschema"] -[tool.black] +[tool.ruff] line-length = 88 +select = ['F', 'E', 'W', 'I', 'N', 'D', 'B', 'A', 'COM', 'C4', 'T20', 'Q', 'SIM'] +exclude = ["tests", "docs"] + [tool.tox] legacy_tox_ini = """ @@ -107,23 +105,7 @@ directory = 'cover' [tool.pytest.ini_options] minversion = "6.0" -addopts = '--cov --cov-report=html --cov-report term-missing --flake8 --mypy --black' -flake8-max-line-length = 88 -flake8-extensions =[ - 'flake8-docstrings', - 'flake8-comprehensions', - 'flake8-import-order', - 'flake8-bugbear', - 'flake8-blind-except', - 'flake8-builtins', - 'flake8-logging-format', - 'flake8-black' -] -flake8-ignore = [ - '* E203', - 'tests/* ALL', - 'docs/* ALL' - ] +addopts = '--cov --cov-report=html --cov-report term-missing --ruff --mypy --black' filterwarnings =[ 'ignore::pytest.PytestDeprecationWarning', 'ignore::DeprecationWarning',