Skip to content

Commit

Permalink
Black format all code
Browse files Browse the repository at this point in the history
  • Loading branch information
jvrsantacruz committed May 28, 2024
1 parent c1ae1b8 commit 66fc7ed
Show file tree
Hide file tree
Showing 5 changed files with 378 additions and 334 deletions.
168 changes: 78 additions & 90 deletions confight.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,27 @@
from __future__ import print_function

import io
import os
import sys
import argparse
import glob
import io
import itertools
import json
import logging
import argparse
import itertools
import os
import sys
from collections import OrderedDict

try:
from ConfigParser import ConfigParser

# Monkey patch python 2.7 version to avoid deprecation warnings
setattr(ConfigParser, 'read_file', getattr(ConfigParser, 'readfp'))
setattr(ConfigParser, "read_file", getattr(ConfigParser, "readfp"))
except ImportError:
from configparser import ConfigParser, ExtendedInterpolation

import toml

__version__ = '2.0.0-2'
logger = logging.getLogger('confight')
__version__ = "2.0.0-2"
logger = logging.getLogger("confight")


def load_user_app(name, extension="toml", user_prefix=None, **kwargs):
Expand All @@ -35,10 +37,10 @@ def load_user_app(name, extension="toml", user_prefix=None, **kwargs):
:returns: Single dict with all the loaded config
"""
if user_prefix is None:
user_prefix = os.path.join('~/.config', name)
filename = 'config.{ext}'.format(ext=extension)
kwargs.setdefault('user_file_path', os.path.join(user_prefix, filename))
kwargs.setdefault('user_dir_path', os.path.join(user_prefix, 'conf.d'))
user_prefix = os.path.join("~/.config", name)
filename = "config.{ext}".format(ext=extension)
kwargs.setdefault("user_file_path", os.path.join(user_prefix, filename))
kwargs.setdefault("user_dir_path", os.path.join(user_prefix, "conf.d"))
return load_app(name, extension, **kwargs)


Expand All @@ -53,15 +55,22 @@ def load_app(name, extension="toml", prefix=None, **kwargs):
:returns: Single dict with all the loaded config
"""
if prefix is None:
prefix = os.path.join('/etc', name)
filename = 'config.{ext}'.format(ext=extension)
kwargs.setdefault('file_path', os.path.join(prefix, filename))
kwargs.setdefault('dir_path', os.path.join(prefix, 'conf.d'))
prefix = os.path.join("/etc", name)
filename = "config.{ext}".format(ext=extension)
kwargs.setdefault("file_path", os.path.join(prefix, filename))
kwargs.setdefault("dir_path", os.path.join(prefix, "conf.d"))
return load_app_paths(extension=extension, **kwargs)


def load_app_paths(file_path=None, dir_path=None, user_file_path=None,
user_dir_path=None, default=None, paths=None, **kwargs):
def load_app_paths(
file_path=None,
dir_path=None,
user_file_path=None,
user_dir_path=None,
default=None,
paths=None,
**kwargs
):
"""Parse and merge user and app config files
User config will have precedence
Expand All @@ -70,19 +79,18 @@ def load_app_paths(file_path=None, dir_path=None, user_file_path=None,
:param dir_path: Path to the extension config directory
:param user_file_path: Path to the user base config file
:param user_dir_path: Path to the user base config file
:param default: Path to be preppended as the default config file embedded
:param default: Path to be prepended as the default config file embedded
in the app
:param paths: Extra paths to add to the parsing after the defaults
:param force_extension: only read files with given extension.
:returns: Single dict with all the loaded config
"""
files = [default, file_path, dir_path, user_file_path, user_dir_path]
files += (paths or [])
files += paths or []
return load_paths([path for path in files if path], **kwargs)


def load_paths(paths, finder=None, extension=None,
force_extension=False, **kwargs):
def load_paths(paths, finder=None, extension=None, force_extension=False, **kwargs):
"""Parse and merge config in path and directories
:param finder: Finder function(dir_path) returning ordered list of paths
Expand All @@ -92,7 +100,7 @@ def load_paths(paths, finder=None, extension=None,
finder = find if finder is None else finder
files = itertools.chain.from_iterable(finder(path) for path in paths)
if extension and force_extension:
files = (path for path in files if path.endswith('.' + extension))
files = (path for path in files if path.endswith("." + extension))
return load(files, **kwargs)


Expand All @@ -118,11 +126,11 @@ def parse(path, format=None):
:returns: dict with the parsed contents
"""
format = format_from_path(path) if format is None else format
logger.info('Parsing %r config file from %r', format, path)
logger.info("Parsing %r config file from %r", format, path)
if format not in FORMATS:
raise ValueError('Unknown format {} for file {}'.format(format, path))
raise ValueError("Unknown format {} for file {}".format(format, path))
loader = FORMAT_LOADERS[format]
with io.open(path, 'r', encoding='utf8') as stream:
with io.open(path, "r", encoding="utf8") as stream:
return loader(stream)


Expand All @@ -136,12 +144,10 @@ def merge(configs):
:param configs: List of parsed config dicts in order
:returns: dict with the merged resulting config
"""
logger.debug('Merging config data %r', configs)
logger.debug("Merging config data %r", configs)
result = OrderedDict()
# No OrderedSets available
keys = OrderedDict(
(key, None) for config in configs for key in config
)
keys = OrderedDict((key, None) for config in configs for key in config)
for key in keys:
values = [config[key] for config in configs if key in config]
merges = [v for v in values if isinstance(v, dict)]
Expand All @@ -164,51 +170,48 @@ def find(path):
return []
if os.path.isfile(path):
return [path]
return sorted(glob.glob(os.path.join(path, '*')))
return sorted(glob.glob(os.path.join(path, "*")))


def check_access(path):
"""Return whether a config file or directory can be read"""
if not path:
return False
elif not os.path.exists(path):
logger.debug('Could not find %r', path)
logger.debug("Could not find %r", path)
return False
elif not os.access(path, os.R_OK):
logger.error('Could not read %r', path)
logger.error("Could not read %r", path)
return False
elif os.path.isdir(path) and not os.access(path, os.X_OK):
logger.error('Could not list directory %r', path)
logger.error("Could not list directory %r", path)
return False
elif os.path.isfile(path) and os.access(path, os.X_OK):
logger.warning('Config file %r has exec permissions', path)
logger.warning("Config file %r has exec permissions", path)
return True


def load_ini(stream):
if 'ExtendedInterpolation' in globals():
if "ExtendedInterpolation" in globals():
parser = ConfigParser(interpolation=ExtendedInterpolation())
else:
parser = ConfigParser()
parser.read_file(stream)
return {
section: OrderedDict(parser.items(section))
for section in parser.sections()
}
return {section: OrderedDict(parser.items(section)) for section in parser.sections()}


FORMATS = ('toml', 'ini', 'json')
FORMATS = ("toml", "ini", "json")
FORMAT_EXTENSIONS = {
'js': 'json',
'json': 'json',
'toml': 'toml',
'ini': 'ini',
'cfg': 'ini',
"js": "json",
"json": "json",
"toml": "toml",
"ini": "ini",
"cfg": "ini",
}
FORMAT_LOADERS = {
'json': lambda *args: json.load(*args, object_pairs_hook=OrderedDict),
'toml': lambda *args: toml.load(*args, _dict=OrderedDict),
'ini': load_ini
"json": lambda *args: json.load(*args, object_pairs_hook=OrderedDict),
"toml": lambda *args: toml.load(*args, _dict=OrderedDict),
"ini": load_ini,
}


Expand All @@ -218,51 +221,43 @@ def load_ini(stream):
except ImportError:
pass
else:

def load_yaml(stream):
yaml = YAML(typ="rt")
return yaml.load(stream)

FORMATS = FORMATS + ('yaml',)
FORMAT_EXTENSIONS.update({
'yml': 'yaml',
'yaml': 'yaml'
})
FORMAT_LOADERS.update({
'yaml': load_yaml
})
FORMATS = FORMATS + ("yaml",)
FORMAT_EXTENSIONS.update({"yml": "yaml", "yaml": "yaml"})
FORMAT_LOADERS.update({"yaml": load_yaml})

# Optional dependency HCL
try:
import hcl
except ImportError:
pass
else:

def load_hcl(stream):
return hcl.load(stream)

FORMATS = FORMATS + ('hcl',)
FORMAT_EXTENSIONS.update({
'hcl': 'hcl'
})
FORMAT_LOADERS.update({
'hcl': load_hcl
})
FORMATS = FORMATS + ("hcl",)
FORMAT_EXTENSIONS.update({"hcl": "hcl"})
FORMAT_LOADERS.update({"hcl": load_hcl})


def format_from_path(path):
"""Get file format from a given path based on exension"""
"""Get file format from a given path based on extension"""
ext = os.path.splitext(path)[1][1:] # extension without dot
format = FORMAT_EXTENSIONS.get(ext)
if not format:
raise ValueError(
'Unknown format extension {!r} for {!r}'.format(ext, path)
)
raise ValueError("Unknown format extension {!r} for {!r}".format(ext, path))
return format


def get_version():
import pkg_resources
return 'confight ' + pkg_resources.get_distribution('confight').version

return "confight " + pkg_resources.get_distribution("confight").version


def cli_configure_logging(args):
Expand All @@ -272,40 +267,33 @@ def cli_configure_logging(args):

def cli_show(args):
"""Load config and show it"""
config = load_user_app(
args.name, prefix=args.prefix, user_prefix=args.user_prefix
)
print(toml.dumps(config), end='')
config = load_user_app(args.name, prefix=args.prefix, user_prefix=args.user_prefix)
print(toml.dumps(config), end="")


def cli():
LOG_LEVELS = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
parser = argparse.ArgumentParser(
description='One simple way of parsing configs'
)
parser.add_argument('--version', action='version', version=get_version())
LOG_LEVELS = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
parser = argparse.ArgumentParser(description="One simple way of parsing configs")
parser.add_argument("--version", action="version", version=get_version())
parser.add_argument(
'-v', '--verbose', choices=LOG_LEVELS, default='ERROR',
help='Logging level default: ERROR'
)
subparsers = parser.add_subparsers(title='subcommands', dest='command')
show_parser = subparsers.add_parser('show')
show_parser.add_argument('name', help='Name of the application')
show_parser.add_argument('--prefix', help='Base for default paths')
show_parser.add_argument(
'--user-prefix', help='Base for default user paths'
"-v", "--verbose", choices=LOG_LEVELS, default="ERROR", help="Logging level default: ERROR"
)
subparsers = parser.add_subparsers(title="subcommands", dest="command")
show_parser = subparsers.add_parser("show")
show_parser.add_argument("name", help="Name of the application")
show_parser.add_argument("--prefix", help="Base for default paths")
show_parser.add_argument("--user-prefix", help="Base for default user paths")

args = parser.parse_args()
cli_configure_logging(args)
# Use callbacks, parser.set_defaults(func=) does not work in Python3.3
callbacks = {
'show': cli_show,
"show": cli_show,
None: lambda args: parser.print_help(file=sys.stderr),
}
try:
callbacks[args.command](args)
except Exception as error:
log = logger.exception if args.verbose == 'DEBUG' else logger.error
log('Error: %s', error)
log = logger.exception if args.verbose == "DEBUG" else logger.error
log("Error: %s", error)
sys.exit(1)
1 change: 1 addition & 0 deletions dev-requirements.txt
Original file line number Diff line number Diff line change
@@ -1,2 +1,3 @@
black
pyHamcrest
pytest
37 changes: 19 additions & 18 deletions setup.py
Original file line number Diff line number Diff line change
@@ -1,36 +1,37 @@
import io
import re

from setuptools import setup


def get_version_from_debian_changelog():
try:
with io.open('debian/changelog', encoding='utf8') as stream:
return re.search(r'\((.+)\)', next(stream)).group(1)
with io.open("debian/changelog", encoding="utf8") as stream:
return re.search(r"\((.+)\)", next(stream)).group(1)
except Exception:
return '0.0.1'
return "0.0.1"


setup(
name='confight',
name="confight",
version=get_version_from_debian_changelog(),
description='Common config loading for Python and the command line',
license='MIT',
author='Avature',
author_email='platform@avature.net',
url='https://github.com/avature/confight',
keywords='config configuration droplets toml json ini yaml',
long_description=io.open('README.md', encoding='utf8').read(),
long_description_content_type='text/markdown',
py_modules=['confight'],
install_requires=io.open('requirements.txt').read().splitlines(),
description="Common config loading for Python and the command line",
license="MIT",
author="Avature",
author_email="platform@avature.net",
url="https://github.com/avature/confight",
keywords="config configuration droplets toml json ini yaml",
long_description=io.open("README.md", encoding="utf8").read(),
long_description_content_type="text/markdown",
py_modules=["confight"],
install_requires=io.open("requirements.txt").read().splitlines(),
extras_require={
'yaml': ["ruamel.yaml>=0.18.0"],
'hcl': ["pyhcl"],
"yaml": ["ruamel.yaml>=0.18.0"],
"hcl": ["pyhcl"],
},
entry_points={
'console_scripts': [
'confight = confight:cli',
"console_scripts": [
"confight = confight:cli",
]
},
classifiers=[
Expand Down
Loading

0 comments on commit 66fc7ed

Please sign in to comment.