mirror of
https://github.com/dtomlinson91/panaetius.git
synced 2025-12-22 13:05:45 +00:00
Compare commits
23 Commits
manual-hea
...
feature/to
| Author | SHA1 | Date | |
|---|---|---|---|
| 525107ad63 | |||
| d604179cbf | |||
| 31fe9b1afc | |||
| 78b86967e7 | |||
| ad840e6b27 | |||
| 9299a12eb6 | |||
| f73a6d2441 | |||
| 4ae4eb085c | |||
| c318045258 | |||
| 035d2b4bef | |||
| b47170070a | |||
| 957ce56a4c | |||
| 4b51a040ce | |||
| e4ae3f0363 | |||
| 1300974a04 | |||
| 517fe974c6 | |||
| 8dfae28832 | |||
| 2c0735fedf | |||
| b9721f6ee4 | |||
| 2885ec8903 | |||
| c1ce2651ac | |||
| 6301cfaae5 | |||
| aec21f30c6 |
20
.vscode/settings.json
vendored
20
.vscode/settings.json
vendored
@@ -3,5 +3,21 @@
|
|||||||
"python.linting.prospectorEnabled": true,
|
"python.linting.prospectorEnabled": true,
|
||||||
"python.linting.enabled": true,
|
"python.linting.enabled": true,
|
||||||
"python.pythonPath": ".venv/bin/python",
|
"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"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
9
panaetius/__init__.py
Normal file
9
panaetius/__init__.py
Normal 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
171
panaetius/config.py
Normal 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
13
panaetius/exceptions.py
Normal 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
43
panaetius/library.py
Normal 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
141
panaetius/logging.py
Normal 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
1423
poetry.lock
generated
File diff suppressed because it is too large
Load Diff
117
prospector.yaml
Normal file
117
prospector.yaml
Normal 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
|
||||||
@@ -25,22 +25,15 @@ classifiers = [
|
|||||||
[tool.poetry.dependencies]
|
[tool.poetry.dependencies]
|
||||||
python = "^3.7"
|
python = "^3.7"
|
||||||
toml = "^0.10.0"
|
toml = "^0.10.0"
|
||||||
pylite = "^0.1.0"
|
PyYAML = "^6.0"
|
||||||
|
|
||||||
[tool.poetry.dev-dependencies]
|
[tool.poetry.dev-dependencies]
|
||||||
pytest = "^3.0"
|
prospector = {extras = ["with_bandit", "with_mypy"], version = "^1.5.1"}
|
||||||
autopep8 = "^1.4"
|
types-toml = "^0.10.1"
|
||||||
pudb = "^2019.2"
|
pytest = "^6.2.5"
|
||||||
McCabe = "^0.6.1"
|
pytest-datadir = "^1.3.1"
|
||||||
YAPF = "^0.29.0"
|
pytest-xdist = "^2.4.0"
|
||||||
pydocstyle = "^5.0"
|
coverage = "^6.0.2"
|
||||||
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"
|
|
||||||
|
|
||||||
[build-system]
|
[build-system]
|
||||||
requires = ["poetry>=0.12"]
|
requires = ["poetry>=0.12"]
|
||||||
|
|||||||
3
pytest.ini
Normal file
3
pytest.ini
Normal file
@@ -0,0 +1,3 @@
|
|||||||
|
; ; parallel tests with pytest-xdist
|
||||||
|
; [pytest]
|
||||||
|
; addopts=-n4
|
||||||
43
rewrite.todo
Normal file
43
rewrite.todo
Normal 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)
|
||||||
@@ -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
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
__version__ = '1.0.2'
|
|
||||||
@@ -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 = []
|
|
||||||
@@ -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)
|
|
||||||
@@ -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]
|
|
||||||
@@ -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'
|
|
||||||
@@ -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()
|
|
||||||
@@ -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)
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
__header__ = 'panaetius_test'
|
|
||||||
23
tests/conftest.py
Normal file
23
tests/conftest.py
Normal 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]},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
10
tests/data/without_logging/panaetius_testing/config.toml
Normal file
10
tests/data/without_logging/panaetius_testing/config.toml
Normal 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] }
|
||||||
9
tests/data/without_logging/panaetius_testing/config.yml
Normal file
9
tests/data/without_logging/panaetius_testing/config.yml
Normal 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
57
tests/scratchpad.py
Normal 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))))
|
||||||
@@ -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
256
tests/test_config.py
Normal 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
13
tests/test_library.py
Normal 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
34
tests/test_logging.py
Normal 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)
|
||||||
@@ -1,5 +0,0 @@
|
|||||||
from panaetius import __version__
|
|
||||||
|
|
||||||
|
|
||||||
def test_version():
|
|
||||||
assert __version__ == '0.1.0'
|
|
||||||
Reference in New Issue
Block a user