NikolajDanger/meow_base
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added files missed by last commit

Browse Source
This commit is contained in:
PatchOfScotland
2022-12-02 13:15:17 +01:00
parent 07ceca0061
commit 00b5449089
5 changed files with 240 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
core/correctness/validation.py
View File

@ -1,5 +1,8 @@
def check_input(variable, expected_type, or_none=False):
from typing import Any, _SpecialForm
def check_input(variable:Any, expected_type:type, alt_types:list[type]=[],
or_none:bool=False)->None:
"""
Checks if a given variable is of the expected type. Raises TypeError or
ValueError as appropriate if any issues are encountered.
@ -8,27 +11,35 @@ def check_input(variable, expected_type, or_none=False):
:param expected_type: (type) expected type of the provided variable
:param alt_types: (optional)(list) additional types that are also
acceptable
:param or_none: (optional) boolean of if the variable can be unset.
Default value is False.
:return: No return.
"""
type_list = [expected_type]
type_list = type_list + alt_types
if not or_none:
if not isinstance(variable, expected_type):
if expected_type != Any \
and type(variable) not in type_list:
raise TypeError(
'Expected type was %s, got %s'
% (expected_type, type(variable))
)
else:
if not isinstance(variable, expected_type) \
if expected_type != Any \
and not type(variable) not in type_list \
and not isinstance(variable, type(None)):
raise TypeError(
'Expected type was %s or None, got %s'
% (expected_type, type(variable))
)
def valid_string(variable, valid_chars):
def valid_string(variable:str, valid_chars:str, min_length:int=1)->None:
"""
Checks that all characters in a given string are present in a provided
list of characters. Will raise an ValueError if unexpected character is
@ -38,14 +49,50 @@ def valid_string(variable, valid_chars):
:param valid_chars: (str) collection of valid characters.
:param min_length: (int) minimum length of variable.
:return: No return.
"""
check_input(variable, str)
check_input(valid_chars, str)
if len(variable) < min_length:
raise ValueError (
f"String '{variable}' is too short. Minimum length is {min_length}"
)
for char in variable:
if char not in valid_chars:
raise ValueError(
"Invalid character '%s'. Only valid characters are: "
"%s" % (char, valid_chars)
)
def valid_dict(variable:dict[Any, Any], key_type:type, value_type:type,
required_keys:list[Any]=[], optional_keys:list[Any]=[],
strict:bool=True)->None:
check_input(variable, dict)
check_input(key_type, type, alt_types=[_SpecialForm])
check_input(value_type, type, alt_types=[_SpecialForm])
check_input(required_keys, list)
check_input(optional_keys, list)
check_input(strict, bool)
for k, v in variable.items():
if key_type != Any and not isinstance(k, key_type):
raise TypeError(f"Key {k} had unexpected type '{type(k)}' "
f"rather than expected '{key_type}' in dict '{variable}'")
if value_type != Any and not isinstance(v, value_type):
raise TypeError(f"Value {v} had unexpected type '{type(v)}' "
f"rather than expected '{value_type}' in dict '{variable}'")
for rk in required_keys:
if rk not in variable.keys():
raise KeyError(f"Missing required key '{rk}' from dict "
f"'{variable}'")
if strict:
for k in variable.keys():
if k not in required_keys and k not in optional_keys:
raise ValueError(f"Unexpected key '{k}' should not be present "
f"in dict '{variable}'")

8
core/correctness/vars.py
View File

@ -1,4 +1,6 @@
import os
CHAR_LOWERCASE = 'abcdefghijklmnopqrstuvwxyz'
CHAR_UPPERCASE = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
CHAR_NUMERIC = '0123456789'
@ -8,3 +10,9 @@ VALID_NAME_CHARS = CHAR_UPPERCASE + CHAR_LOWERCASE + CHAR_NUMERIC + "_-"
VALID_RECIPE_NAME_CHARS = VALID_NAME_CHARS
VALID_PATTERN_NAME_CHARS = VALID_NAME_CHARS
VALID_RULE_NAME_CHARS = VALID_NAME_CHARS
VALID_VARIABLE_NAME_CHARS = CHAR_UPPERCASE + CHAR_LOWERCASE + CHAR_NUMERIC + "_"
VALID_JUPYTER_NOTEBOOK_FILENAME_CHARS = VALID_NAME_CHARS + "." + os.path.sep
VALID_JUPYTER_NOTEBOOK_EXTENSIONS = [".ipynb"]
VALID_TRIGGERING_PATH_CHARS = VALID_NAME_CHARS + "." + os.path.sep

151
core/meow.py
View File

@ -5,92 +5,139 @@ import core.correctness.validation
from typing import Any
class BaseRecipe:
name: str
recipe: Any
paramaters: dict[str, Any]
def __init__(self, name:str, recipe:Any, parameters:dict[str,Any]={}):
self.__is_valid_name(name)
name:str
recipe:Any
parameters:dict[str, Any]
requirements:dict[str, Any]
def __init__(self, name:str, recipe:Any, parameters:dict[str,Any]={},
requirements:dict[str,Any]={}):
self._is_valid_name(name)
self.name = name
self.__is_valid_recipe(recipe)
self._is_valid_recipe(recipe)
self.recipe = recipe
self.__is_valid_parameters(parameters)
self.paramaters = parameters
self._is_valid_parameters(parameters)
self.parameters = parameters
self._is_valid_requirements(requirements)
self.requirements = requirements
def __init_subclass__(cls, **kwargs) -> None:
if cls._is_valid_recipe == BaseRecipe._is_valid_recipe:
raise NotImplementedError(
f"Recipe '{cls.__name__}' has not implemented "
"'_is_valid_recipe(self, recipe)' function.")
if cls._is_valid_parameters == BaseRecipe._is_valid_parameters:
raise NotImplementedError(
f"Recipe '{cls.__name__}' has not implemented "
"'_is_valid_parameters(self, parameters)' function.")
if cls._is_valid_requirements == BaseRecipe._is_valid_requirements:
raise NotImplementedError(
f"Recipe '{cls.__name__}' has not implemented "
"'_is_valid_requirements(self, requirements)' function.")
super().__init_subclass__(**kwargs)
def __new__(cls, *args, **kwargs):
if cls is BaseRecipe:
raise TypeError("BaseRecipe may not be instantiated directly")
return object.__new__(cls)
def __is_valid_name(self, name):
def _is_valid_name(self, name:str)->None:
core.correctness.validation.valid_string(
name, core.correctness.vars.VALID_RECIPE_NAME_CHARS)
def __is_valid_recipe(self, recipe):
raise NotImplementedError(
f"Recipe '{self.__class__.__name__}' has not implemented "
"'__is_valid_recipe(self, recipe)' function.")
def _is_valid_recipe(self, recipe:Any)->None:
pass
def _is_valid_parameters(self, parameters:Any)->None:
pass
def _is_valid_requirements(self, requirements:Any)->None:
pass
def __is_valid_parameters(self, parameters):
raise NotImplementedError(
f"Recipe '{self.__class__.__name__}' has not implemented "
"'__is_valid_parameters(self, parameters)' function.")
class BasePattern:
name: str
recipe: BaseRecipe
parameters: dict[str, Any]
outputs: dict[str, Any]
def __init__(self, name:str, recipe:BaseRecipe,
parameters:dict[str,Any]={}, outputs:dict[str,Any]={}):
self.__is_valid_name(name)
name:str
recipe:str
parameters:dict[str, Any]
outputs:dict[str, Any]
def __init__(self, name:str, recipe:str, parameters:dict[str,Any]={},
outputs:dict[str,Any]={}):
self._is_valid_name(name)
self.name = name
self.__is_valid_recipe(recipe)
self._is_valid_recipe(recipe)
self.recipe = recipe
self.__is_valid_parameters(parameters)
self.paramaters = parameters
self.__is_valid_output(outputs)
self._is_valid_parameters(parameters)
self.parameters = parameters
self._is_valid_output(outputs)
self.outputs = outputs
def __init_subclass__(cls, **kwargs) -> None:
if cls._is_valid_recipe == BasePattern._is_valid_recipe:
raise NotImplementedError(
f"Pattern '{cls.__name__}' has not implemented "
"'_is_valid_recipe(self, recipe)' function.")
if cls._is_valid_parameters == BasePattern._is_valid_parameters:
raise NotImplementedError(
f"Pattern '{cls.__name__}' has not implemented "
"'_is_valid_parameters(self, parameters)' function.")
if cls._is_valid_output == BasePattern._is_valid_output:
raise NotImplementedError(
f"Pattern '{cls.__name__}' has not implemented "
"'_is_valid_output(self, outputs)' function.")
super().__init_subclass__(**kwargs)
def __new__(cls, *args, **kwargs):
if cls is BasePattern:
raise TypeError("BasePattern may not be instantiated directly")
return object.__new__(cls)
def __is_valid_name(self, name):
def _is_valid_name(self, name:str)->None:
core.correctness.validation.valid_string(
name, core.correctness.vars.VALID_PATTERN_NAME_CHARS)
def __is_valid_recipe(self, recipe):
raise NotImplementedError(
f"Pattern '{self.__class__.__name__}' has not implemented "
"'__is_valid_recipe(self, recipe)' function.")
def _is_valid_recipe(self, recipe:Any)->None:
pass
def __is_valid_parameters(self, parameters):
raise NotImplementedError(
f"Pattern '{self.__class__.__name__}' has not implemented "
"'__is_valid_parameters(self, parameters)' function.")
def _is_valid_parameters(self, parameters:Any)->None:
pass
def _is_valid_output(self, outputs:Any)->None:
pass
def __is_valid_output(self, outputs):
raise NotImplementedError(
f"Pattern '{self.__class__.__name__}' has not implemented "
"'__is_valid_output(self, outputs)' function.")
class BaseRule:
name: str
patterns: list[BasePattern]
def __init__(self, name:str, patterns:list[BasePattern]):
self.__is_valid_name(name)
name:str
pattern:BasePattern
recipe:BaseRecipe
def __init__(self, name:str, pattern:BasePattern, recipe:BaseRecipe):
self._is_valid_name(name)
self.name = name
self.__is_valid_patterns(patterns)
self.patterns = patterns
self._is_valid_pattern(pattern)
self.pattern = pattern
self._is_valid_recipe(recipe)
self.recipe = recipe
def __new__(cls, *args, **kwargs):
if cls is BaseRule:
raise TypeError("BaseRule may not be instantiated directly")
return object.__new__(cls)
def __is_valid_name(self, name):
def __init_subclass__(cls, **kwargs) -> None:
if cls._is_valid_pattern == BaseRule._is_valid_pattern:
raise NotImplementedError(
f"Rule '{cls.__name__}' has not implemented "
"'_is_valid_pattern(self, pattern)' function.")
if cls._is_valid_recipe == BaseRule._is_valid_recipe:
raise NotImplementedError(
f"Pattern '{cls.__name__}' has not implemented "
"'_is_valid_recipe(self, recipe)' function.")
super().__init_subclass__(**kwargs)
def _is_valid_name(self, name:str)->None:
core.correctness.validation.valid_string(
name, core.correctness.vars.VALID_RULE_NAME_CHARS)
def __is_valid_patterns(self, patterns):
raise NotImplementedError(
f"Rule '{self.__class__.__name__}' has not implemented "
"'__is_valid_patterns(self, patterns)' function.")
def _is_valid_pattern(self, pattern:Any)->None:
pass
def _is_valid_recipe(self, recipe:Any)->None:
pass

41
patterns/FileEventPattern.py
View File

@ -1,5 +1,44 @@
from typing import Any
from core.correctness.validation import check_input, valid_string, valid_dict
from core.correctness.vars import VALID_RECIPE_NAME_CHARS, \
VALID_VARIABLE_NAME_CHARS
from core.meow import BasePattern
class FileEventPattern(BasePattern):
pass
triggering_path:str
triggering_file:str
def __init__(self, name:str, triggering_path:str, recipe:str,
triggering_file:str, parameters:dict[str,Any]={},
outputs:dict[str,Any]={}):
super().__init__(name, recipe, parameters, outputs)
self._is_valid_triggering_path(triggering_path)
self.triggering_path = triggering_path
self._is_valid_triggering_file(triggering_file)
self.triggering_file = triggering_file
def _is_valid_recipe(self, recipe:str)->None:
valid_string(recipe, VALID_RECIPE_NAME_CHARS)
def _is_valid_triggering_path(self, triggering_path:str)->None:
check_input(triggering_path, str)
if len(triggering_path) < 1:
raise ValueError (
f"trigginering path '{triggering_path}' is too short. "
"Minimum length is 1"
)
def _is_valid_triggering_file(self, triggering_file:str)->None:
valid_string(triggering_file, VALID_VARIABLE_NAME_CHARS)
def _is_valid_parameters(self, parameters:dict[str,Any])->None:
valid_dict(parameters, str, Any, strict=False)
for k in parameters.keys():
valid_string(k, VALID_VARIABLE_NAME_CHARS)
def _is_valid_output(self, outputs:dict[str,str])->None:
valid_dict(outputs, str, str, strict=False)
for k in outputs.keys():
valid_string(k, VALID_VARIABLE_NAME_CHARS)

43
recipes/JupyterNotebookRecipe.py
View File

@ -1,5 +1,46 @@
import nbformat
from typing import Any
from core.correctness.validation import check_input, valid_string, valid_dict
from core.correctness.vars import VALID_JUPYTER_NOTEBOOK_FILENAME_CHARS, \
VALID_JUPYTER_NOTEBOOK_EXTENSIONS, VALID_VARIABLE_NAME_CHARS
from core.meow import BaseRecipe
class JupyterNotebookRecipe(BaseRecipe):
pass
source:str
def __init__(self, name:str, recipe:Any, parameters:dict[str,Any]={},
requirements:dict[str,Any]={}, source:str=""):
super().__init__(name, recipe, parameters, requirements)
self._is_valid_source(source)
self.source = source
def _is_valid_source(self, source:str)->None:
valid_string(
source, VALID_JUPYTER_NOTEBOOK_FILENAME_CHARS, min_length=0)
if not source:
return
matched = False
for i in VALID_JUPYTER_NOTEBOOK_EXTENSIONS:
if source.endswith(i):
matched = True
if not matched:
raise ValueError(f"source '{source}' does not end with a valid "
"jupyter notebook extension.")
def _is_valid_recipe(self, recipe:dict[str,Any])->None:
check_input(recipe, dict)
nbformat.validate(recipe)
def _is_valid_parameters(self, parameters:dict[str,Any])->None:
valid_dict(parameters, str, Any, strict=False)
for k in parameters.keys():
valid_string(k, VALID_VARIABLE_NAME_CHARS)
def _is_valid_requirements(self, requirements:dict[str,Any])->None:
valid_dict(requirements, str, Any, strict=False)
for k in requirements.keys():
valid_string(k, VALID_VARIABLE_NAME_CHARS)