21 Commits

Author SHA1 Message Date
525107ad63 adding a config.yml instead of config.toml 2021-10-19 21:46:41 +01:00
d604179cbf updating todos 2021-10-18 02:36:27 +01:00
31fe9b1afc removing old source code 2021-10-18 02:32:15 +01:00
78b86967e7 Merge branch 'feature/rewrite' into develop 2021-10-18 02:31:35 +01:00
ad840e6b27 adding latest testing + docstrings 2021-10-18 02:31:17 +01:00
9299a12eb6 adding latest tests 2021-10-18 01:03:49 +01:00
f73a6d2441 adding latest tests 2021-10-18 00:12:20 +01:00
4ae4eb085c adding initial tests 2021-10-17 06:51:47 +01:00
c318045258 adding latest working 2021-10-17 06:51:41 +01:00
035d2b4bef adding latest working 2021-10-16 16:36:15 +01:00
b47170070a adding latest not working on bool 2021-10-16 15:29:18 +01:00
957ce56a4c adding latest working config complete 2021-10-16 06:05:58 +01:00
4b51a040ce adding latest working 2021-10-16 05:55:04 +01:00
e4ae3f0363 adding latest working 2021-10-16 05:44:39 +01:00
1300974a04 updating prospector.yaml 2021-10-16 05:25:07 +01:00
517fe974c6 updating todo 2021-10-16 05:25:00 +01:00
8dfae28832 adding latest working 2021-10-16 05:24:56 +01:00
2c0735fedf adding latest working 2021-10-16 04:47:32 +01:00
b9721f6ee4 adding latest 2021-10-16 02:16:53 +01:00
2885ec8903 adding latest 2021-10-15 04:14:25 +01:00
c1ce2651ac updating dependencies 2021-10-11 00:06:36 +01:00
30 changed files with 1585 additions and 1536 deletions

20
.vscode/settings.json vendored
View File

@@ -3,5 +3,21 @@
"python.linting.prospectorEnabled": true,
"python.linting.enabled": true,
"python.pythonPath": ".venv/bin/python",
"restructuredtext.confPath": "${workspaceFolder}/docs/source"
}
"restructuredtext.confPath": "${workspaceFolder}/docs/source",
"peacock.color": "#307E6A",
"workbench.colorCustomizations": {
"editorGroup.border": "#3ea389",
"panel.border": "#3ea389",
"sash.hoverBorder": "#3ea389",
"sideBar.border": "#3ea389",
"statusBar.background": "#307e6a",
"statusBar.foreground": "#e7e7e7",
"statusBarItem.hoverBackground": "#3ea389",
"statusBarItem.remoteBackground": "#307e6a",
"statusBarItem.remoteForeground": "#e7e7e7",
"titleBar.activeBackground": "#307e6a",
"titleBar.activeForeground": "#e7e7e7",
"titleBar.inactiveBackground": "#307e6a99",
"titleBar.inactiveForeground": "#e7e7e799"
}
}

3
TODO Normal file
View File

@@ -0,0 +1,3 @@
Todo:
☐ Item

9
panaetius/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
"""
panaetius - a utility library to read variables and provide convenient logging.
Author: Daniel Tomlinson (dtomlinson@panaetius.co.uk)
"""
from panaetius.config import Config
from panaetius.library import set_config
from panaetius.logging import set_logger, SimpleLogger, AdvancedLogger, CustomLogger

171
panaetius/config.py Normal file
View File

@@ -0,0 +1,171 @@
"""
Access variables from a config file or an environment variable.
This module defines the `Config` class to interact and read variables from either a
`config.yml` or an environment variable.
"""
from __future__ import annotations
import ast
import os
import pathlib
from typing import Any
# import toml
import yaml
from panaetius.exceptions import KeyErrorTooDeepException, InvalidPythonException
class Config:
"""The configuration class to access variables."""
def __init__(self, header_variable: str, config_path: str = "") -> None:
"""
Create a Config object to set and access variables.
Args:
header_variable (str): Your header variable name.
config_path (str, optional): The path where the header directory is stored.
Defaults to `~/.config`.
Example:
A header of `data_analysis` with a config_path of `~/myapps` will define
a config file in `~/myapps/data_analysis/config.yml`.
"""
self.header_variable = header_variable
self.config_path = (
pathlib.Path(config_path)
if config_path
else pathlib.Path.home() / ".config"
)
self._missing_config = self._check_config_file_exists()
# default logging options
self.logging_path: str | None = None
self.logging_rotate_bytes: int = 0
self.logging_backup_count: int = 0
@property
def config(self) -> dict:
"""
Return the contents of the config file. If missing returns an empty dictionary.
Returns:
dict: The contents of the `.yml` loaded as a python dictionary.
"""
config_file_location = self.config_path / self.header_variable / "config.yml"
try:
with open(config_file_location, "r", encoding="utf-8") as config_file:
# return dict(toml.load(config_file))
return dict(yaml.load(stream=config_file, Loader=yaml.SafeLoader))
except FileNotFoundError:
return {}
def get_value(self, key: str, default: Any) -> Any:
"""
Get the value of a variable from the key name.
The key can either be one (`value`) or two (`data.value`) levels deep.
A key of `value` (with a header of `data_analysis`) would refer to a
`config.yml` of:
```
[data_analysis]
value = "some value"
```
or an environment variable of `DATA_ANALYSIS_VALUE="'some value'"`.
A key of `data.value` would refer to a `config.yml` of:
```
[data_analysis.data]
value = "some value"
```
or an environment variable of `DATA_ANALYSIS_DATA_VALUE="'some value'"`.
Args:
key (str): The key of the variable.
default (Any): The default value if the key cannot be found in the config
file, or an environment variable.
Returns:
Any: The value of the variable.
"""
env_key = f"{self.header_variable.upper()}_{key.upper().replace('.', '_')}"
if not self._missing_config:
# look in the config file
return self._get_config_value(env_key, key, default)
# no config file, look for env vars
return self._get_env_value(env_key, default)
def _check_config_file_exists(self) -> bool:
config_file_location = self.config_path / self.header_variable / "config.yml"
try:
with open(config_file_location, "r", encoding="utf-8"):
return False
except FileNotFoundError:
return True
def _get_config_value(self, env_key: str, key: str, default: Any) -> Any:
try:
# look under top header
# REVIEW: could this be auto handled for a key of arbitrary length?
# check for env variable and have it take priority
value = os.environ.get(env_key.replace("-", "_"))
if value is not None:
return self.__get_config_value_env_var_override(value)
if len(key.split(".")) > 2:
raise KeyErrorTooDeepException(
f"Your key of {key} can only be 2 levels deep maximum. "
f"You have {len(key.split('.'))}"
)
if len(key.split(".")) == 1:
return self.__get_config_value_key_split_once(key)
if len(key.split(".")) == 2:
return self.__get_config_value_key_split_twice(key)
raise KeyError()
except (KeyError, TypeError):
if value is None:
return self.__get_config_value_missing_key_value_is_none(default)
# if env var is present, load it
return self.__get_config_value_missing_key_value_is_not_none(value)
def __get_config_value_key_split_once(self, key: str) -> Any:
name = key.lower()
return self.config[self.header_variable][name]
def __get_config_value_key_split_twice(self, key: str) -> Any:
section, name = key.lower().split(".")
return self.config[self.header_variable][section][name]
def __get_config_value_missing_key_value_is_none(self, default: Any) -> Any:
return self.__load_default_value(default)
def __get_config_value_missing_key_value_is_not_none(self, value: str) -> Any:
return self.__load_value(value)
def __get_config_value_env_var_override(self, value: str) -> Any:
return self.__load_value(value)
def _get_env_value(self, env_key: str, default: Any) -> Any: # noqa
# look for an environment variable, fallback to default
value = os.environ.get(env_key.replace("-", "_"))
if value is None:
return self.__load_default_value(default)
return self.__load_value(value)
def __load_value(self, value: str) -> Any: # noqa
try:
return ast.literal_eval(value)
except (ValueError, SyntaxError):
raise InvalidPythonException(f"{value} is not valid Python.") # noqa
def __load_default_value(self, default: Any) -> Any: # noqa
return default

13
panaetius/exceptions.py Normal file
View File

@@ -0,0 +1,13 @@
"""Exceptions for the module."""
class KeyErrorTooDeepException(Exception):
pass
class LoggingDirectoryDoesNotExistException(Exception):
pass
class InvalidPythonException(Exception):
pass

43
panaetius/library.py Normal file
View File

@@ -0,0 +1,43 @@
"""Module to provide functionality when interacting with variables."""
from __future__ import annotations
from typing import Any
from panaetius import Config
def set_config(
config_inst: Config,
key: str,
default: Any = None,
) -> None:
"""
Define a variable to be read from a `config.toml` or an environment variable.
Args:
config_inst (Config): The instance of the `Config` class.
key (str): The key of the variable.
default (Any, optional): The default value if the key cannot be found in the config
file, or an environment variable. Defaults to None.
Example:
`set_config(CONFIG, "value", default=[1, 2])` would look for a
`config.toml` with the following structure (with `CONFIG` having a header of
`data_analysis`):
```
[data_analysis]
value = "some value"
```
Or an environment variable of `DATA_ANALYSIS_VALUE="'some value'"`.
If found, this value can be access with `CONFIG.value` which would return
`some_value`.
If neither the environment variable nor the `config.toml` are present, the
default of `[1, 2]` would be returned instead.
"""
config_var = key.lower().replace(".", "_")
setattr(config_inst, config_var, config_inst.get_value(key, default))

141
panaetius/logging.py Normal file
View File

@@ -0,0 +1,141 @@
"""Module to define a convenient logger instance with json formatted output."""
from __future__ import annotations
from abc import ABCMeta, abstractmethod
import logging
from logging.handlers import RotatingFileHandler
import pathlib
import sys
from panaetius import Config
from panaetius.library import set_config
from panaetius.exceptions import LoggingDirectoryDoesNotExistException
def set_logger(config_inst: Config, logging_format_inst: LoggingData) -> logging.Logger:
"""
Set and return a `logging.Logger` instance for quick logging.
`logging_format_inst` should be an instance of either SimpleLogger, AdvancedLogger,
or CustomLogger.
SimpleLogger and AdvancedLogger define a logging format and a logging level info.
CustomLogger defines a logging level info and should have a logging format passed
in.
Logging to a file is defined by a `logging.path` key set on `Config`. This path
should exist as it will not be created.
Args:
config_inst (Config): The instance of the `Config` class.
logging_format_inst (LoggingData): The instance of the `LoggingData` class.
Raises:
LoggingDirectoryDoesNotExistException: If the logging directory specified does
not exist.
Returns:
logging.Logger: An configured instance of `logging.Logger` ready to be used.
Example:
```
logger = set_logger(CONFIG, SimpleLogger())
logger.info("some logging message")
```
Would create a logging output of:
```
{
"time": "2021-10-18 02:26:24,037",
"logging_level":"INFO",
"message": "some logging message"
}
```
"""
logger = logging.getLogger(config_inst.header_variable)
log_handler_sys = logging.StreamHandler(sys.stdout)
# configure file handler
if config_inst.logging_path is not None:
logging_file = (
pathlib.Path(config_inst.logging_path)
/ config_inst.header_variable
/ f"{config_inst.header_variable}.log"
).expanduser()
if not logging_file.parents[0].exists():
raise LoggingDirectoryDoesNotExistException()
if config_inst.logging_rotate_bytes == 0:
set_config(config_inst, "logging.rotate_bytes", 512000)
if config_inst.logging_backup_count == 0:
set_config(config_inst, "logging.backup_count", 3)
log_handler_file = RotatingFileHandler(
str(logging_file),
"a",
config_inst.logging_rotate_bytes,
config_inst.logging_backup_count,
)
log_handler_file.setFormatter(logging.Formatter(logging_format_inst.format))
logger.addHandler(log_handler_file)
# configure stdout handler
log_handler_sys.setFormatter(logging.Formatter(logging_format_inst.format))
logger.addHandler(log_handler_sys)
logger.setLevel(logging_format_inst.logging_level)
return logger
class LoggingData(metaclass=ABCMeta):
@property
@abstractmethod
def format(self) -> str:
pass
@abstractmethod
def __init__(self, logging_level: str):
self.logging_level = logging_level
class SimpleLogger(LoggingData):
@property
def format(self) -> str:
return str(
'{\n\t"time": "%(asctime)s",\n\t"logging_level":'
'"%(levelname)s",\n\t"message": "%(message)s"\n}',
)
def __init__(self, logging_level: str = "INFO"):
self.logging_level = logging_level
class AdvancedLogger(LoggingData):
@property
def format(self) -> str:
return str(
'{\n\t"time": "%(asctime)s",\n\t"file_name": "%(filename)s",'
'\n\t"module": "%(module)s",\n\t"function":"%(funcName)s",\n\t'
'"line_number": "%(lineno)s",\n\t"logging_level":'
'"%(levelname)s",\n\t"message": "%(message)s"\n}',
)
def __init__(self, logging_level: str = "INFO"):
self.logging_level = logging_level
class CustomLogger(LoggingData):
@property
def format(self) -> str:
return str(self._format)
def __init__(self, logging_format: str, logging_level: str = "INFO"):
self.logging_level = logging_level
self._format = logging_format

1423
poetry.lock generated

File diff suppressed because it is too large Load Diff

117
prospector.yaml Normal file
View File

@@ -0,0 +1,117 @@
output-format: vscode
doc-warnings: true
strictness: none
ignore-patterns:
- (^|/)\..+
# https://pylint.pycqa.org/en/latest/technical_reference/features.html
pylint:
run: true
disable:
# disables TODO warnings
- fixme
# !doc docstrings
# - missing-module-docstring
# - missing-class-docstring
# - missing-function-docstring
# ! doc end of docstrings
# disables warnings about abstract methods not overridden
- abstract-method
# used when an ancestor class method has an __init__ method which is not called by a derived class.
- super-init-not-called
# either all return statements in a function should return an expression, or none of them should.
# - inconsistent-return-statements
# Used when an expression that is not a function call is assigned to nothing. Probably something else was intended.
# - expression-not-assigned
# Used when a line is longer than a given number of characters.
- line-too-long
enable:
options:
max-locals: 15
max-returns: 6
max-branches: 12
max-statements: 50
max-parents: 7
max-attributes: 20
min-public-methods: 0
max-public-methods: 25
max-module-lines: 1000
max-line-length: 88
max-args: 8
mccabe:
run: true
options:
max-complexity: 10
# https://pep8.readthedocs.io/en/release-1.7.x/intro.html#error-codes
pep8:
run: true
options:
max-line-length: 88
single-line-if-stmt: n
disable:
# line too long
- E501
pyroma:
run: false
disable:
- PYR19
- PYR16
# https://pep257.readthedocs.io/en/latest/error_codes.html
# http://www.pydocstyle.org/en/6.1.1/error_codes.html
pep257:
disable:
# !doc docstrings
# Missing docstring in __init__
# - D107
# Missing docstring in public module
# - D100
# Missing docstring in public class
# - D101
# Missing docstring in public method
# - D102
# Missing docstring in public function
# - D103
# Multi-line docstring summary should start at the second line
# - D213
# First word of the docstring should not be This
# - D404
# DEFAULT IGNORES
# 1 blank line required before class docstring
- D203
# Multi-line docstring summary should start at the first line
- D212
# !doc end of docstrings
# Section name should end with a newline
- D406
# Missing dashed underline after section
- D407
# Missing blank line after last section
- D413
# https://flake8.pycqa.org/en/latest/user/error-codes.html
pyflakes:
disable:
# module imported but unused
- F401
dodgy:
run: true
bandit:
run: true
# options:
# ignore assert warning
# - B101
mypy:
run: true
options:
# https://mypy.readthedocs.io/en/stable/running_mypy.html#missing-type-hints-for-third-party-library
ignore-missing-imports: true
# https://mypy.readthedocs.io/en/stable/running_mypy.html#following-imports
follow-imports: normal

View File

@@ -25,22 +25,15 @@ classifiers = [
[tool.poetry.dependencies]
python = "^3.7"
toml = "^0.10.0"
pylite = "^0.1.0"
PyYAML = "^6.0"
[tool.poetry.dev-dependencies]
pytest = "^3.0"
autopep8 = "^1.4"
pudb = "^2019.2"
McCabe = "^0.6.1"
YAPF = "^0.29.0"
pydocstyle = "^5.0"
Pyflakes = "^2.1"
Rope = "^0.16.0"
python-language-server = "^0.31.4"
pycodestyle = "^2.5"
sphinx = "^2.3"
sphinx_rtd_theme = "^0.4.3"
prospector = "^1.3.0"
prospector = {extras = ["with_bandit", "with_mypy"], version = "^1.5.1"}
types-toml = "^0.10.1"
pytest = "^6.2.5"
pytest-datadir = "^1.3.1"
pytest-xdist = "^2.4.0"
coverage = "^6.0.2"
[build-system]
requires = ["poetry>=0.12"]

3
pytest.ini Normal file
View File

@@ -0,0 +1,3 @@
; ; parallel tests with pytest-xdist
; [pytest]
; addopts=-n4

43
rewrite.todo Normal file
View File

@@ -0,0 +1,43 @@
Documentation:
☐ Rewrite documentation using `mkdocs` and using `.md`.
☐ Update the metadata in the `pyproject.toml`.
☐ Create a new `Readme.md` and remove the `.rst`.
Misc:
☐ Use the python runner to build the docs & run the tests (including coverage html)
coverage run -m pytest && coverage report && coverage html
☐ document this in trilium
☐ Bump the version to release 2.0
Archive:
✔ Handle if a bool is passed in as a default @done(21-10-16 05:25) @project(Coding.No Config File)
✔ Handle if a bool is passed in as a default @done(21-10-16 05:25) @project(Coding.Config File)
✔ Create SimpleLogger, AdvancedLogger, CustomLogger classes @done(21-10-16 16:22) @project(Coding.Logging)
✔ Logging path should take by default the config path unless overwritten? @done(21-10-16 23:49) @project(Coding.Logging)
✔ Check logging path + config path are valid, if not raise error. @done(21-10-18 00:04) @project(Coding.Errors)
✔ Add tests for these. @done(21-10-18 00:04) @project(Coding.Errors)
✔ Check for a key > 2 levels, raise custom error, write test @done(21-10-17 23:30) @project(Coding.Errors)
✔ Check all functions and annotations. @done(21-10-18 01:07) @project(Coding.Linting)
✔ Write the docstrings for public functions/methods. @done(21-10-18 02:29) @project(Coding.Docstrings)
✔ When both a config file and a env var is found, use the env var. @done(21-10-18 00:38) @project(Coding.Functionality)
✔ If loading from a default, don't covert to TOML @done(21-10-17 20:33) @project(Tests.Bugfixes)
✔ Env Vars should be given as python objects @done(21-10-17 20:33) @project(Tests.Bugfixes)
The string or node provided may only consist of the following Python literal structures: strings, bytes, numbers, tuples, lists, dicts, sets, booleans, and None.
use ast.literal_eval()
https://docs.python.org/3/library/ast.html#ast.literal_eval
✔ Test default config path set to "~/.config" @done(21-10-17 17:25) @project(Tests.__init__)
✔ Test config path is set when passed in @done(21-10-17 17:25) @project(Tests.__init__)
✔ Check testing config file is returned as dict @done(21-10-17 17:25) @project(Tests.config property)
✔ Check _self.missing_config and empty dict is returned @done(21-10-17 17:25) @project(Tests.config property)
✔ Arrays & tables loaded correctly from config file @done(21-10-17 20:34) @project(Tests.get_value.config_file)
✔ test when key length is 1 the value is returned @done(21-10-17 18:55) @project(Tests.get_value.config_file)
✔ test when key length is 2 the value is returned @done(21-10-17 18:55) @project(Tests.get_value.config_file)
✔ test when key not found and no env var default is loaded @done(21-10-17 19:01) @project(Tests.get_value.config_file)
✔ test bool's are properly converted @done(21-10-17 19:01) @project(Tests.get_value.config_file)
✔ test when key not found and env var is set value is loaded @done(21-10-17 20:43) @project(Tests.get_value.config_file)
✔ check if env key is missing the default is read in @done(21-10-17 20:55) @project(Tests.get_value.env_var)
✔ check if env key is present the values are read in @done(21-10-17 22:24) @project(Tests.get_value.env_var)
✔ parametrise a test to read in values form env vars and they're set correctly @done(21-10-17 22:24) @project(Tests.get_value.env_var)
✔ test that the env var is valid python @done(21-10-18 01:03) @project(Tests.get_value.env_var)
✔ test set_config works @done(21-10-17 23:29) @project(Tests.library)

View File

@@ -1,6 +0,0 @@
from panaetius.config_inst import CONFIG
from .config import Config
from .library import set_config
from panaetius.header import __header__
import panaetius.logging
from panaetius.logging import logger as logger

View File

@@ -1 +0,0 @@
__version__ = '1.0.2'

View File

@@ -1,207 +0,0 @@
from typing import Callable, Union
import os
import toml
from panaetius.library import export
from panaetius.header import __header__
from panaetius.db import Mask
# __all__ = ['Config']
@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:`~panaetius.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.
Mask : panaetius.db.Mask
Class to mask values in a config file.
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`.
path : str
Path to config file
Parameters
----------
path : str
Path to config file
"""
def __init__(self, path: str, header: str = __header__) -> None:
"""
See :class:`~panaetius.config.Config` for parameters.
"""
self.path = os.path.expanduser(path)
self.header = header
self.deferred_messages = []
self.config_file = self.read_config(path)
self.module_name = self.header.lower()
self.Mask = Mask
def read_config(self, path: str, write: bool = False) -> 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`
header : str
Header to overwrite if using in a module.
Example: ``path = '~/.config/panaetius'``
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::
[panaetius]
[panaetius.foo]
foo = bar
Returns a dict:
``{'panaetius' : {foo: {'foo': 'bar'}}}``
"""
path += 'config.toml' if path[-1] == '/' else '/config.toml'
path = os.path.expanduser(path)
if not write:
try:
with open(path, 'r+') as config_file:
config_file = toml.load(config_file)
self.defer_log(f'Config file found at {path}')
return config_file
except FileNotFoundError:
self.defer_log(f'Config file not found at {path}')
else:
try:
with open(path, 'w+') as config_file:
config_file = toml.load(config_file)
self.defer_log(f'Config file found at {path}')
return config_file
except FileNotFoundError:
self.defer_log(f'Config file not found at {path}')
def get(
self,
key: str,
default: str = None,
cast: Callable = None,
mask: bool = False,
) -> Union[str, None]:
"""Retrives the config variable from either the `config.toml` or an
environment variable. Will default to the default value if nothing
is found
Parameters
----------
key : str
Key to the configuration variable. Should be in the form
`panaetius.variable` or `panaetius.header.variable`.
When loaded, it will be accessable at
`Config.panaetius_variable` or
`Config.panaetius_header_variable`.
default : str, optional
The default value if nothing is found. Defaults to `None`.
cast : Callable, optional
The type of the variable. E.g `int` or `float`. Should reference
the type object and not as string. Defaults to `None`.
Returns
-------
Any
Will return the config variable if found, or the default.
"""
env_key = f"{self.header.upper()}_{key.upper().replace('.', '_')}"
try:
# look in the config.toml
if len(key.split('.')) == 2:
# look for subsections
# print(mask)
if mask:
# print('mask', key)
value = self.Mask(
self.path, self.config_file, key
).get_value()
else:
# print('no-mask')
section, name = key.lower().split('.')
value = self.config_file[self.module_name][section][name]
self.defer_log(f'{env_key} found in config.toml')
else:
# print('valueerror')
# look under top level module self.header
# key = f'{self.module_name}.key'
if mask:
# key = f'{self.header}.{key}'
# print(f'mask key={key}')
value = self.Mask(
self.path, self.config_file, key
).get_value()
else:
name = key.lower()
value = self.config_file[self.module_name][name]
self.defer_log(f'{env_key} found in config.toml')
# finally:
try:
# return if found in config.toml
return cast(value) if cast else value
except UnboundLocalError:
# pass if nothing was found
# print('unbound error')
pass
except KeyError:
# print('key error')
self.defer_log(f'{env_key} not found in config.toml')
except TypeError:
# print('type error')
self.defer_log(f'{env_key} not found in config.toml')
# look for an environment variable
value = os.environ.get(env_key.replace("-", "_"))
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

@@ -1,11 +0,0 @@
import os
from panaetius.header import __header__
from panaetius.config import Config
DEFAULT_CONFIG_PATH = f'~/.config/{__header__.lower()}'
CONFIG_PATH = os.environ.get(
f'{__header__.upper()}_CONFIG_PATH', DEFAULT_CONFIG_PATH
)
CONFIG = Config(CONFIG_PATH)

View File

@@ -1,256 +0,0 @@
from os import path, urandom
import hashlib
from typing import Tuple
import toml
import io
from pylite.simplite import Pylite
from panaetius.header import __header__ as __header__
import panaetius
class Mask:
"""Class to handle masking sensitive values in a config file
Attributes
----------
config_contents : dict
A dict containing the contents of the config file.
config_path : str
The path to the config file.
config_var : str
The key corresponding to the config entry.
database : Pylite
A Pylite instance for the datbase.
entry : str
The result from the config file. Could either be a hash or the raw
value.
header : str
The __header__ which denotes where the config file is stored.
name : str
The key of the entry in the config file.
result : str
The value of the entry in the config file.
table_name : str
The sqlite table name. Defaults to the __header__ value.
"""
@property
def hash(self):
"""Property to determine the hash of a config entry.
Returns
-------
bytes
The hash as a bytes object.
"""
try:
if not self._hash_exists:
pass
except AttributeError:
self._hash = hashlib.pbkdf2_hmac(
'sha256',
self.entry[self.name].encode('utf-8'),
self.salt,
100000,
dklen=12,
)
self._hash_exists = True
finally:
return self._hash
@property
def salt(self):
"""Property to detemine a random salt to use in creation of the hash.
Returns
-------
bytes
The salt as a bytes object.
"""
self._salt = urandom(32)
return self._salt
@staticmethod
def as_string(obj: bytes) -> str:
"""Static method to return a string from a bytes object.
Parameters
----------
obj : bytes
Bytes object to be converted to a string.
Returns
-------
str
The bytes object as a string.
"""
return bytes.hex(obj)
@staticmethod
def fromhex(obj: str) -> bytes:
"""Static method to create a bytes object from a string.
Parameters
----------
obj : str
String object to be converted to bytes.
Returns
-------
bytes
The string object as bytes.
"""
return bytes.fromhex(obj)
@staticmethod
def _from_key(config_var) -> Tuple[str, str]:
try:
header, name = config_var.split('.')
except ValueError:
header = ''
name = config_var
return (header, name)
def __init__(
self, config_path: str, config_contents: dict, config_var: str
):
"""Summary
See :class:`~Mask` for parameters.
"""
self.table: str = __header__
self.config_path = config_path
self.config_contents = config_contents
self.config_var = config_var.replace('.', '_')
self.header = self._from_key(config_var)[0]
self.name = self._from_key(config_var)[1]
try:
# If value is under a subsection
self.entry = self.config_contents[self.table][self.header]
except KeyError:
# If value is under the main header
self.entry = self.config_contents[self.table]
def _get_database_file(self):
self.database = self.config_path
self.database += (
f'.{self.table}.db'
if self.config_path[-1] == '/'
else f'/.{self.table}.db'
)
self.database = path.expanduser(self.database)
return self
def _open_database(self):
self.database = Pylite(self.database)
def _get_table(self):
tables = [i[0] for i in self.database.get_tables()]
if self.table not in tables:
# panaetius.logger.debug(
# 'Table not present in the database;'
# f'creating the table {self.table} now'
# )
self.database.add_table(
f'{self.table}',
Name='text',
Hash='text',
Salt='text',
Value='text',
)
else:
# panaetius.logger.debug('Table already exists in the database')
pass
self.table_name = self.table
def _check_entries(self):
var = self.database.get_items(self.table, f'Name="{self.config_var}"')
if len(var) == 0:
return False
else:
return True
def _insert_entries(self):
self.database.insert(
self.table,
self.config_var,
self.as_string(self.hash),
self.as_string(self.salt),
self.entry[self.name],
)
def _update_entries_in_db(self):
self.database.remove(self.table, f'Name="{self.config_var}"')
self._insert_entries()
def _run_query(self, query: str):
cur = self.database.db.cursor()
cur.execute(query)
self.database.db.commit()
self.result = cur.fetchall()
return self
def _get_all_items(self, where_clause: str = None):
if where_clause is not None:
self.result = self.database.get_items(self.table, where_clause)
else:
self.result = self.database.get_items(self.table)
return self
def _process(self):
if not self._check_entries():
# panaetius.logger.debug('does not exist')
self._insert_entries()
self._update_entries_in_config()
self._get_all_items()
# panaetius.logger.debug(f'returning: {self.result[0][3]}')
return self.entry[self.name]
else:
self._get_all_items(f'Name="{self.config_var}"')
if self.result[0][1] == self.entry[self.name]:
# panaetius.logger.debug('exists and hash matches')
# panaetius.logger.debug(f'returning: {self.result[0][3]}')
return self.result
else:
# panaetius.logger.debug('exists and hash doesnt match')
# panaetius.logger.debug(
# f'file_hash={self.entry[self.name]}, {self.result[0][1]}'
# )
self._update_entries_in_db()
self._update_entries_in_config()
self._get_all_items(f'Name="{self.config_var}"')
# panaetius.logger.debug(f'returning: {self.result[0][3]}')
return self.entry[self.name]
def _open_config_file(self) -> io.TextIOWrapper:
self.config_path += (
'/config.toml' if self.config_path[-1] != '/' else 'config.toml'
)
c = open(path.expanduser(self.config_path), 'w')
return c
def _update_entries_in_config(self):
self.entry.update({self.name: self.as_string(self.hash)})
# panaetius.logger.debug(self.config_contents)
# panaetius.logger.debug(self.entry)
c = self._open_config_file()
toml.dump(self.config_contents, c)
c.close()
def get_value(self):
"""Get the true value from the database if it exists, create if it'
' doesn't exist or update if the hash has changed.
Returns
-------
str
The result from the database.
"""
# print(f'key in db {self.config_var}')
self._get_database_file()
self._open_database()
self._get_table()
self._process()
return self.result[0][3]

View File

@@ -1,26 +0,0 @@
import os
from importlib import util
__path = os.getcwd()
try:
__spec = util.spec_from_file_location(
'__header__', f'{os.getcwd()}/__header__.py'
)
__header__ = util.module_from_spec(__spec)
__spec.loader.exec_module(__header__)
__header__ = __header__.__header__
except FileNotFoundError:
try:
venv = os.environ.get('VIRTUAL_ENV').split('/')[-1]
__header__ = venv
except AttributeError:
print(
f'Cannot find a __header__.py file in {os.getcwd()} containing the'
' __header__ value of your project name and you are not working'
' from a virtual environment. Either make sure this file '
'exists and the value is set or create and work from a virtual '
'environment and try again. \n The __header__ value has been '
'set to the default of panaetius.'
)
__header__ = 'panaetius'

View File

@@ -1,112 +0,0 @@
from __future__ import annotations
import sys
from typing import Any, TypeVar, Type, TYPE_CHECKING, Union, List
import ast
if TYPE_CHECKING:
import logging
config_inst_t = TypeVar('config_inst_t', bound='panaetius.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,
check: Union[None, List] = None,
mask: bool = False,
) -> None:
"""Sets the config variable on the instance of a class.
Parameters
----------
config_inst : Type[config_inst_t]
Instance of the :class:`~panaetius.config.Config` class.
key : str
The key referencing the config variable.
default : str, optional
The default value.
mask : bool, optional
Boolean to indiciate if a value in the `config.toml` should be masked.
If this is set to True then the first time the variable is read from
the config file the value will be replaced with a hash. Any time that
value is then read the hash will be compared to the one stored and if
they match the true value will be returned. This is stored in a sqlite
`.db` next to the config file and is hidden by default. If the hash
provided doesn't match the default behaviour is to update the `.db`
with the new value and hash the value again. If you delete the
database file then you will need to set the value again in the
`config.toml`.
cast : Any, optional
The type of the variable.
check : Union[None, List], optional
Type of object to check against. This is useful if you want to use TOML
to define a list, but want to make sure that a string representation
of a list will be loaded properly if it set as an environment variable.
Example:
*config.toml* has the following attribute set::
[package.users]
auth = ['user1', 'user2']
If set as an environment variable you can pass this list as a string
and set :code:`check=list`::
Environment variable:
PACKAGE_USERS_AUTH = "['user1', 'user2']"
Usage in code::
set_config(CONFIG, 'users.auth', check=list)
"""
config_var = key.lower().replace('.', '_')
if check is None:
setattr(
config_inst, config_var, config_inst.get(key, default, cast, mask)
)
else:
if type(config_inst.get(key, default, cast, mask)) is not check:
if check is list:
var = ast.literal_eval(
config_inst.get(key, default, cast, mask)
)
setattr(config_inst, config_var, var)
else:
setattr(
config_inst,
config_var,
config_inst.get(key, default, cast, mask),
)
# Create function to print cached logged messages and reset
def process_cached_logs(
config_inst: Type[config_inst_t], logger: logging.Logger
):
"""Prints the cached messages from :class:`~panaetius.config.Config`
and resets the cache.
Parameters
----------
config_inst : Type[config_inst_t]
Instance of :class:`~panaetius.config.Config`.
logger : logging.Logger
Instance of the logger.
"""
for msg in config_inst.deferred_messages:
logger.info(msg)
config_inst.reset_log()

View File

@@ -1,54 +0,0 @@
import logging
from logging.handlers import RotatingFileHandler
import os
import sys
import panaetius
from panaetius import CONFIG as CONFIG
from panaetius import __header__ as __header__
from panaetius import set_config as set_config
panaetius.set_config(CONFIG, 'logging.path')
panaetius.set_config(
CONFIG,
'logging.format',
'{\n\t"time": "%(asctime)s",\n\t"file_name": "%(filename)s",'
'\n\t"module": "%(module)s",\n\t"function":"%(funcName)s",\n\t'
'"line_number": "%(lineno)s",\n\t"logging_level":'
'"%(levelname)s",\n\t"message": "%(message)s"\n}',
cast=str,
)
set_config(CONFIG, 'logging.level', 'INFO')
# Logging Configuration
logger = logging.getLogger(__header__)
loghandler_sys = logging.StreamHandler(sys.stdout)
# Checking if log path is set
if CONFIG.logging_path:
CONFIG.logging_path += (
f'{__header__}.log'
if CONFIG.logging_path[-1] == '/'
else f'/{__header__}.log'
)
# Set default log file options
set_config(CONFIG, 'logging.backup_count', 3, int)
set_config(CONFIG, 'logging.rotate_bytes', 512000, int)
# Configure file handler
loghandler_file = RotatingFileHandler(
os.path.expanduser(CONFIG.logging_path),
'a',
CONFIG.logging_rotate_bytes,
CONFIG.logging_backup_count,
)
# Add to file formatter
loghandler_file.setFormatter(logging.Formatter(CONFIG.logging_format))
logger.addHandler(loghandler_file)
# Configure and add to stdout formatter
loghandler_sys.setFormatter(logging.Formatter(CONFIG.logging_format))
logger.addHandler(loghandler_sys)
logger.setLevel(CONFIG.logging_level)

View File

@@ -1 +0,0 @@
__header__ = 'panaetius_test'

23
tests/conftest.py Normal file
View File

@@ -0,0 +1,23 @@
import pytest
@pytest.fixture()
def header():
return "panaetius_testing"
@pytest.fixture()
def testing_config_contents():
return {
"panaetius_testing": {
"some_top_string": "some_top_value",
"second": {
"some_second_string": "some_second_value",
"some_second_int": 1,
"some_second_float": 1.0,
"some_second_list": ["some", "second", "value"],
"some_second_table": {"first": ["some", "first", "value"]},
"some_second_table_bools": {"bool": [True, False]},
},
}
}

View File

@@ -0,0 +1,10 @@
[panaetius_testing]
some_top_string = "some_top_value"
[panaetius_testing.second]
some_second_string = "some_second_value"
some_second_int = 1
some_second_float = 1.0
some_second_list = ["some", "second", "value"]
some_second_table = { "first" = ["some", "first", "value"] }
some_second_table_bools = { "bool" = [true, false] }

View File

@@ -0,0 +1,9 @@
panaetius_testing:
some_top_string: some_top_value
second:
some_second_string: some_second_value
some_second_int: 1
some_second_float: 1.0
some_second_list: ["some", "second", "value"]
some_second_table: { "first": ["some", "first", "value"] }
some_second_table_bools: { "bool": [true, false] }

57
tests/scratchpad.py Normal file
View File

@@ -0,0 +1,57 @@
import os
from panaetius import Config, set_config, set_logger, SimpleLogger
from panaetius.logging import AdvancedLogger
if __name__ == "__main__":
os.environ["PANAETIUS_TEST_PATH"] = '"/usr/local"'
os.environ["PANAETIUS_TEST_BOOL"] = "True"
# print(os.environ.get("PANAETIUS_TEST_PATH"))
# os.environ[
# "PANAETIUS_TEST_TOML_POINTS"
# ] = "[ { x = 1, y = 2, z = 3 }, { x = 7, y = 8, z = 9 }, { x = 2, y = 4, z = 8 }]"
os.environ["PANAETIUS_TEST_NOC_PATH"] = '"/usr/locals"'
os.environ["PANAETIUS_TEST_NOC_FLOAT"] = "2.0"
os.environ["PANAETIUS_TEST_NOC_BOOL"] = "True"
os.environ["PANAETIUS_TEST_NOC_EMBEDDED_PATH"] = '"/usr/local"'
os.environ["PANAETIUS_TEST_NOC_EMBEDDED_FLOAT"] = "2.0"
os.environ["PANAETIUS_TEST_NOC_EMBEDDED_BOOL"] = "True"
c = Config("panaetius_test")
# c = Config("panaetius_test_noc")
set_config(c, key="toml.points")
set_config(c, key="path", default="some path")
set_config(c, key="top", default="some top")
set_config(c, key="logging.path")
set_config(c, key="nonexistent.item", default="some nonexistent item")
set_config(c, key="nonexistent.item")
set_config(c, key="toml.points_config")
set_config(c, key="float")
set_config(c, key="float_str", default="2.0")
set_config(c, key="bool")
set_config(c, key="noexistbool", default=False)
set_config(c, key="middle.middle")
# set_config(c, key="path")
# set_config(c, key="float")
# set_config(c, key="bool")
# set_config(c, key="noexiststr", default="2.0")
# set_config(c, key="noexistfloat", default=2.0)
# set_config(c, key="noexistbool", default=False)
set_config(c, key="embedded.path")
set_config(c, key="embedded.float")
set_config(c, key="embedded.bool")
set_config(c, key="embedded.noexiststr", default="2.0")
set_config(c, key="embedded.noexistfloat", default=2.0)
set_config(c, key="embedded.noexistbool", default=False)
# logger = set_logger(c, SimpleLogger())
logger = set_logger(c, SimpleLogger(logging_level="DEBUG"))
logger.info("some logging message")
logger.debug("debugging message")
# for i in dir(c):
# logger.debug(i + ": " + str(getattr(c, i)) + " - " + str(type(getattr(c, i))))

View File

@@ -1,33 +0,0 @@
import panaetius
# from panaetius import CONFIG as CONFIG
# from panaetius import logger as logger
print(panaetius.__header__)
panaetius.set_config(panaetius.CONFIG, 'logging.level')
# print(panaetius.CONFIG.logging_format)
print(panaetius.CONFIG.logging_path)
print(panaetius.config_inst.CONFIG_PATH)
# panaetius.logger.info('test event')
panaetius.logger.info('setting foo.bar value')
panaetius.set_config(panaetius.CONFIG, 'foo.bar', mask=True)
panaetius.logger.info(f'foo.bar set to {panaetius.CONFIG.foo_bar}')
# print((panaetius.CONFIG.path))
# print(panaetius.CONFIG.logging_level)
panaetius.set_config(panaetius.CONFIG, 'test', mask=True)
panaetius.logger.info(f'test_root={panaetius.CONFIG.test}')
print(panaetius.CONFIG.config_file)
# for i in panaetius.CONFIG.deferred_messages:
# panaetius.logger.debug(i)
panaetius.logger.info('some logging message')

256
tests/test_config.py Normal file
View File

@@ -0,0 +1,256 @@
import os
import pathlib
from uuid import uuid4
import pytest
import panaetius
from panaetius.exceptions import InvalidPythonException, KeyErrorTooDeepException
# test config paths
def test_default_config_path_set(header):
# act
config = panaetius.Config(header)
# assert
assert str(config.config_path) == str(pathlib.Path.home() / ".config")
def test_user_config_path_set(header, shared_datadir):
# arrange
config_path = str(shared_datadir / "without_logging")
# act
config = panaetius.Config(header, config_path)
# assert
assert str(config.config_path) == config_path
# test config files
def test_config_file_exists(header, shared_datadir):
# arrange
config_path = str(shared_datadir / "without_logging")
# act
config = panaetius.Config(header, config_path)
_ = config.config
# assert
assert config._missing_config is False
def test_config_file_contents_read_success(
header, shared_datadir, testing_config_contents
):
# arrange
config_path = str(shared_datadir / "without_logging")
# act
config = panaetius.Config(header, config_path)
config_contents = config.config
# assert
assert config_contents == testing_config_contents
@pytest.mark.parametrize(
"set_config_key,get_config_key,expected_value",
[
("some_top_string", "some_top_string", "some_top_value"),
("second.some_second_string", "second_some_second_string", "some_second_value"),
(
"second.some_second_list",
"second_some_second_list",
["some", "second", "value"],
),
(
"second.some_second_table",
"second_some_second_table",
{"first": ["some", "first", "value"]},
),
(
"second.some_second_table_bools",
"second_some_second_table_bools",
{"bool": [True, False]},
),
],
)
def test_get_value_from_key(
set_config_key, get_config_key, expected_value, header, shared_datadir
):
"""
Test the following:
- keys are read from top level key
- keys are read from two level key
- inline arrays are read correctly
- inline tables are read correctly
- inline tables & arrays read bools correctly
"""
# arrange
config_path = str(shared_datadir / "without_logging")
config = panaetius.Config(header, config_path)
panaetius.set_config(config, set_config_key)
# act
config_value = getattr(config, get_config_key)
# assert
assert config_value == expected_value
def test_get_value_environment_var_override(header, shared_datadir):
# arrange
os.environ[f"{header.upper()}_SOME_TOP_STRING"] = '"some_overridden_value"'
config_path = str(shared_datadir / "without_logging")
config = panaetius.Config(header, config_path)
panaetius.set_config(config, "some_top_string")
# act
config_value = getattr(config, "some_top_string")
# assert
assert config_value == "some_overridden_value"
# cleanup
del os.environ[f"{header.upper()}_SOME_TOP_STRING"]
def test_key_level_too_deep(header, shared_datadir):
# arrange
config_path = str(shared_datadir / "without_logging")
config = panaetius.Config(header, config_path)
key = "a.key.too.deep"
# act
with pytest.raises(KeyErrorTooDeepException) as key_error_too_deep:
panaetius.set_config(config, key)
# assert
assert (
str(key_error_too_deep.value)
== f"Your key of {key} can only be 2 levels deep maximum. "
f"You have 4"
)
def test_get_value_missing_key_from_default(header, shared_datadir):
# arrange
config_path = str(shared_datadir / "without_logging")
config = panaetius.Config(header, config_path)
panaetius.set_config(
config,
"missing.key_from_default",
default=["some", "default", "value", 1.0, True],
)
# act
default_value = getattr(config, "missing_key_from_default")
# assert
assert default_value == ["some", "default", "value", 1.0, True]
def test_get_value_missing_key_from_env(header, shared_datadir):
# arrange
os.environ[f"{header.upper()}_MISSING_KEY"] = '"some missing key"'
config_path = str(shared_datadir / "without_logging")
config = panaetius.Config(header, config_path)
panaetius.set_config(config, "missing_key")
# act
value_from_key = getattr(config, "missing_key")
# assert
assert value_from_key == "some missing key"
# cleanup
del os.environ[f"{header.upper()}_MISSING_KEY"]
# test env vars
def test_config_file_does_not_exist(header, shared_datadir):
# arrange
config_path = str(shared_datadir / "nonexistent_folder")
# act
config = panaetius.Config(header, config_path)
config_contents = config.config
# assert
assert config._missing_config is True
assert config_contents == {}
def test_missing_config_read_from_default(header, shared_datadir):
# arrange
config_path = str(shared_datadir / "nonexistent_folder")
# act
config = panaetius.Config(header, config_path)
panaetius.set_config(config, "missing.key_read_from_default", default=True)
# assert
assert getattr(config, "missing_key_read_from_default") is True
@pytest.mark.parametrize(
"env_value,expected_value",
[
('"a missing string"', "a missing string"),
("1", 1),
("1.0", 1.0),
("True", True),
(
'["an", "array", "of", "items", 1, True]',
["an", "array", "of", "items", 1, True],
),
(
'{"an": "array", "of": "items", "1": True}',
{"an": "array", "of": "items", "1": True},
),
],
)
def test_missing_config_read_from_env_var(
env_value, expected_value, header, shared_datadir
):
# arrange
config_path = str(shared_datadir / str(uuid4()))
os.environ[f"{header.upper()}_MISSING_KEY_READ_FROM_ENV_VAR"] = env_value
# act
config = panaetius.Config(header, config_path)
panaetius.set_config(config, "missing.key_read_from_env_var")
# assert
assert getattr(config, "missing_key_read_from_env_var") == expected_value
# cleanup
del os.environ[f"{header.upper()}_MISSING_KEY_READ_FROM_ENV_VAR"]
def test_missing_config_read_from_env_var_invalid_python(header):
# arrange
os.environ[f"{header.upper()}_INVALID_PYTHON"] = "a string without quotes"
config = panaetius.Config(header)
# act
with pytest.raises(InvalidPythonException) as invalid_python_exception:
panaetius.set_config(config, "invalid_python")
# assert
assert (
str(invalid_python_exception.value)
== "a string without quotes is not valid Python."
)
# cleanup
del os.environ[f"{header.upper()}_INVALID_PYTHON"]

13
tests/test_library.py Normal file
View File

@@ -0,0 +1,13 @@
import panaetius
def test_set_config(header, shared_datadir):
# arrange
config_path = str(shared_datadir / "without_logging")
# act
config = panaetius.Config(header, config_path)
panaetius.set_config(config, "some_top_string")
# assert
assert getattr(config, "some_top_string") == "some_top_value"

34
tests/test_logging.py Normal file
View File

@@ -0,0 +1,34 @@
import logging
from uuid import uuid4
import pytest
from panaetius import set_logger, SimpleLogger, Config, set_config
from panaetius.exceptions import LoggingDirectoryDoesNotExistException
def test_logging_directory_does_not_exist(header, shared_datadir):
# arrange
config = Config(header)
logging_path = str(shared_datadir / str(uuid4()))
set_config(config, "logging.path", default=str(logging_path))
# act
with pytest.raises(LoggingDirectoryDoesNotExistException) as logging_exception:
_ = set_logger(config, SimpleLogger())
# assert
assert str(logging_exception.value) == ""
def test_logging_directory_does_exist(header, shared_datadir):
# arrange
config = Config(header)
logging_path = str(shared_datadir / "without_logging")
set_config(config, "logging.path", default=str(logging_path))
# act
logger = set_logger(config, SimpleLogger())
# assert
assert isinstance(logger, logging.Logger)

View File

@@ -1,5 +0,0 @@
from panaetius import __version__
def test_version():
assert __version__ == '0.1.0'