adding read-from-config boilerplate

This commit is contained in:
2019-12-02 00:23:00 +00:00
parent 24e42f47b9
commit e2f07a0d9f
4 changed files with 276 additions and 0 deletions

View File

@@ -0,0 +1,43 @@
"""Caching utilities."""
class cachedproperty:
"""A decorator for caching a property's result.
Similar to `property`, but the wrapped method's result is cached
on the instance. This is achieved by setting an entry in the object's
instance dictionary with the same name as the property. When the name
is later accessed, the value in the instance dictionary takes precedence
over the (non-data descriptor) property.
This is useful for implementing lazy-loaded properties.
The cache can be invalidated via `delattr()`, or by modifying `__dict__`
directly. It will be repopulated on next access.
.. versionadded:: 6.3.0
"""
def __init__(self, func, doc=None):
"""Initialize the descriptor."""
self.func = self.__wrapped__ = func
if doc is None:
doc = func.__doc__
self.__doc__ = doc
def __get__(self, obj, objtype=None):
"""Implement descriptor getter.
Calculate the property's value and then store it in the
associated object's instance dictionary.
"""
if obj is None:
return self
value = obj.__dict__[self.func.__name__] = self.func(obj)
return value
def __repr__(self):
"""Return repr(self)."""
return "<%s %s>" % (self.__class__.__name__, self.func)

View File

@@ -0,0 +1,44 @@
from __future__ import annotations
from .config.config import Config
from .library import set_config
from .__header__ import __header__
import logging
from logging.handlers import RotatingFileHandler
import os
import sys
# Load User Defined Config
DEFAULT_CONFIG_PATH = f'~/.config/{__header__.lower()}'
CONFIG_PATH = os.environ.get(f'{__header__}_CONFIG_PATH', DEFAULT_CONFIG_PATH)
CONFIG = Config(CONFIG_PATH)
# Logging Configuration
logger = logging.getLogger(__header__)
set_config(CONFIG, 'logging.path')
set_config(
CONFIG,
'logging.format',
'%(asctime)s - %(module)s:%(lineno)s - %(levelname)s - %(message)s',
)
set_config(CONFIG, 'logging.level', 'INFO')
loghandler_sys = logging.StreamHandler(sys.stdout)
if CONFIG.logging_path:
set_config(CONFIG, 'logging.backup_count', 3, int)
set_config(CONFIG, 'logging.rotate_bytes', 512000, int)
loghandler_file = RotatingFileHandler(
os.path.expanduser(CONFIG.logging_path),
'a',
CONFIG.logging_rotate_bytes,
CONFIG.logging_backup_count
)
loghandler_file.setFormatter(logging.Formatter(CONFIG.logging_format))
logger.addHandler(loghandler_file)
loghandler_sys.setFormatter(logging.Formatter(CONFIG.logging_format))
logger.addHandler(loghandler_sys)
logger.setLevel(CONFIG.logging_level)
for msg in CONFIG.deferred_messages:
logger.info(msg)
CONFIG.reset_log()

View File

@@ -0,0 +1,152 @@
from plex_posters.library import export
from plex_posters.__header__ import __header__ as header
from typing import Callable, Union
import os
import toml
__all__ = []
@export
class Config:
"""Handles the config options for the module and stores config variables
to be shared.
Attributes
----------
config_file : dict
Contains the config options. See
:meth:`~plex_posters.config.config.Config.read_config`
for the data structure.
deferred_messages : list
A list containing the messages to be logged once the logger has been
instantiated.
module_name : str
A string representing the module name. This is added in front of all
envrionment variables and is the title of the `config.toml`.
Parameters
----------
path : str
Path to config file
"""
def __init__(self, path: str) -> None:
"""
See :class:`~plex_posters.config.config.Config` for parameters.
"""
self.config_file = self.read_config(path)
self.module_name = header.lower()
self.deferred_messages = []
def read_config(self, path: str) -> Union[dict, None]:
"""Reads the toml config file from `path` if it exists.
Parameters
----------
path : str
Path to config file. Should not contain `config.toml`
Example: ``path = '~/.config/plex_posters'``
Returns
-------
Union[dict, None]
Returns a dict if the file is found else returns nothing.
The dict contains a key for each header. Each key corresponds to a
dictionary containing a key, value pair for each config under
that header.
Example::
[plex_posters]
[plex_posters.foo]
foo = bar
Returns a dict:
``{'plex_posters' : {foo: {'foo': 'bar'}}}``
"""
path += 'config.toml' if path[-1] == '/' else '/config.toml'
path = os.path.expanduser(path)
try:
with open(path, 'r+') as config_file:
config_file = toml.load(config_file)
return config_file
except FileNotFoundError:
try:
self.defer_log(f'Config file not found at {config_file}')
except UnboundLocalError:
pass
pass
def get(
self, key: str, default: str = None, cast: Callable = None
) -> Union[str, None]:
"""Retrives the config variable from either the `config.toml` or an
environment variable. Will default to the default value if it is
provided.
Parameters
----------
key : str
Key to the configuration variable. Should be in the form
`module.variable` which will be converted to `module_variable`.
default : str, optional
The default value if nothing is found.
cast : Callable, optional
The type of the variable. E.g `int` or `float`. Should reference
the type object and not as string.
Returns
-------
Any
Will return the config variable if found, or the default.
"""
env_key = f"{header}_{key.upper().replace('.', '_')}"
# self.defer_log(self.config_file)
try:
# look in the config.toml
section, name = key.lower().split('.')
value = self.config_file[self.module_name][section][name]
self.defer_log(f'{env_key} found in config.toml')
return cast(value) if cast else value
except KeyError:
self.defer_log(f'{env_key} not found in config.toml')
except TypeError:
pass
# look for an environment variable
value = os.environ.get(env_key)
if value is not None:
self.defer_log(f'{env_key} found in an environment variable')
else:
# fall back to default
self.defer_log(f'{env_key} not found in an environment variable.')
value = default
self.defer_log(f'{env_key} set to default {default}')
return cast(value) if cast else value
def defer_log(self, msg: str) -> None:
"""Populates a list `Config.deferred_messages` with all the events to
be passed to the logger later if required.
Parameters
----------
msg : str
The message to be logged.
"""
self.deferred_messages.append(msg)
def reset_log(self) -> None:
"""Empties the list `Config.deferred_messages`.
"""
del self.deferred_messages
self.deferred_messages = []

View File

@@ -0,0 +1,37 @@
import sys
from typing import Any, TypeVar, Type
config_inst_t = TypeVar('config_inst_t', bound='config.config.Config')
def export(fn: callable) -> callable:
mod = sys.modules[fn.__module__]
if hasattr(mod, '__all__'):
mod.__all__.append(fn.__name__)
else:
mod.__all__ = [fn.__name__]
return fn
def set_config(
config_inst: Type[config_inst_t],
key: str,
default: str = None,
cast: Any = None,
) -> None:
"""Sets the config variable on the instance of a class.
Parameters
----------
config_inst : Type[config_inst_t]
Instance of the config class.
key : str
The key referencing the config variable.
default : str, optional
The default value.
cast : Any, optional
The type of the variable.
"""
config_var = key.lower().replace('.', '_')
setattr(config_inst, config_var, config_inst.get(key, default, cast))