1
0
Fork 0
mirror of https://github.com/yavook/kiwi-scp.git synced 2024-12-25 02:23:01 +00:00

new kiwi.yml format

This commit is contained in:
Jörn-Michael Miehe 2021-10-11 02:58:49 +02:00
parent 5ffb9cc41b
commit c0edd3bcc5
6 changed files with 113 additions and 172 deletions

View file

@ -0,0 +1,23 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="kiwi_next" type="PythonConfigurationType" factoryName="Python" nameIsGenerated="true">
<module name="kiwi-scp" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/kiwi_scp/scripts/kiwi_next.py" />
<option name="PARAMETERS" value="" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
</component>

View file

@ -1,20 +0,0 @@
# local
from .parser import Parser
from .runner import Runner
def verbosity():
# ensure singleton is instantiated: runs subcommand setup routines
_ = Runner()
return Parser().get_args().verbosity
def run():
# pass down
return Runner().run()
__all__ = [
'verbosity',
'run'
]

View file

@ -1,157 +1,79 @@
# system
import copy
import logging
import os
import re
import yaml
from typing import Optional, Dict, List
# local
from ._constants import KIWI_CONF_NAME, HEADER_KIWI_CONF_NAME, DEFAULT_KIWI_CONF_NAME, VERSION_TAG_NAME
import pydantic
class Config:
class _Storage(pydantic.BaseModel):
"""a storage subsection"""
directory: str
class _Project(pydantic.BaseModel):
"""a project subsection"""
name: str
enabled: bool = True
storage: Optional[_Storage]
@pydantic.root_validator(pre=True)
@classmethod
def check_grammar(cls, values):
if isinstance(values, dict):
if "name" in values:
return values
elif len(values) == 1:
name, enabled = list(values.items())[0]
return {"name": name, "enabled": True if enabled is None else enabled}
elif isinstance(values, str):
return {"name": values}
class _Network(pydantic.BaseModel):
"""a network subsection"""
name: str
cidr: str
class Config(pydantic.BaseModel):
"""represents a kiwi.yml"""
__yml_content = {}
__keys = {
'version': "kiwi-scp version to use in this instance",
version: str
shells: Optional[List[str]]
environment: Optional[Dict[str, Optional[str]]]
'runtime:storage': "local directory for service data",
'runtime:shells': "shell preference for working in service containers",
'runtime:env': "common environment for compose yml",
'markers:project': "marker string for project directories",
'markers:disabled': "marker string for disabled projects",
'network:name': "name for local network hub",
'network:cidr': "CIDR block for local network hub",
}
def __key_resolve(self, key):
"""
Resolve nested dictionaries
If __yml_content is {'a': {'b': {'c': "val"}}} and key is 'a:b:c',
this returns a single dict {'c': "val"} and the direct key 'c'
"""
# "a:b:c" => path = ['a', 'b'], key = 'c'
path = key.split(':')
path, key = path[:-1], path[-1]
# resolve path
container = self.__yml_content
for step in path:
container = container[step]
return container, key
def __getitem__(self, key):
"""array-like read access to __yml_content"""
container, key = self.__key_resolve(key)
return container[key]
def __setitem__(self, key, value):
"""array-like write access to __yml_content"""
container, key = self.__key_resolve(key)
container[key] = value
def __str__(self):
"""dump into textual representation"""
# dump yml content
yml_string = yaml.dump(
self.__yml_content,
default_flow_style=False, sort_keys=False
).strip()
# insert newline before every main key
yml_string = re.sub(r'^(\S)', r'\n\1', yml_string, flags=re.MULTILINE)
# load header comment from file
with open(HEADER_KIWI_CONF_NAME, 'r') as stream:
yml_string = stream.read() + yml_string
return yml_string
def _update_from_file(self, filename):
"""return a copy updated using a kiwi.yml file"""
with open(filename, 'r') as stream:
try:
# create copy
result = Config()
result.__yml_content = copy.deepcopy(self.__yml_content)
# read file
logging.debug(f"Reading '{filename}' into '{id(result.__yml_content)}'")
result.__yml_content.update(yaml.safe_load(stream))
return result
except yaml.YAMLError as exc:
logging.error(exc)
def user_query(self, key):
"""query user for new config value"""
# prompt user as per argument
try:
result = input(f"Enter {self.__keys[key]} [{self[key]}] ").strip()
except EOFError:
print()
result = None
# store result if present
if result:
self[key] = result
def save(self):
"""save current yml representation in current directory's kiwi.yml"""
with open(KIWI_CONF_NAME, 'w') as stream:
stream.write(str(self))
stream.write('\n')
class DefaultConfig(Config):
"""Singleton: The default kiwi.yml file"""
__instance = None
projects: Optional[List[_Project]]
storage: _Storage
network: _Network
@pydantic.validator("version")
@classmethod
def get(cls):
if cls.__instance is None:
# create singleton
cls.__instance = cls()._update_from_file(DEFAULT_KIWI_CONF_NAME)
def check_version(cls, value: str) -> str:
if not re.match(r"^[0-9]+(\.[0-9]+(\.[0-9]+)?)?$", value):
raise ValueError
# add version data from separate file (keeps default config cleaner)
with open(VERSION_TAG_NAME, 'r') as stream:
cls.__instance['version'] = stream.read().strip()
# return singleton
return cls.__instance
class LoadedConfig(Config):
"""Singleton collection: kiwi.yml files by path"""
__instances = {}
return value
@pydantic.validator("environment", pre=True)
@classmethod
def get(cls, directory='.'):
if directory not in LoadedConfig.__instances:
# create singleton for new path
result = DefaultConfig.get()
def unify_env(cls, value) -> Optional[Dict[str, Optional[str]]]:
if isinstance(value, dict):
return value
elif isinstance(value, list):
result: Dict[str, Optional[str]] = {}
for item in value:
idx = item.find("=")
if idx == -1:
key, value = item, None
else:
key, value = item[:idx], item[idx + 1:]
# update with that dir's kiwi.yml
try:
result = result._update_from_file(os.path.join(directory, KIWI_CONF_NAME))
except FileNotFoundError:
logging.info(f"No '{KIWI_CONF_NAME}' found at '{directory}'. Using defaults.")
result[key] = value
LoadedConfig.__instances[directory] = result
# return singleton
return LoadedConfig.__instances[directory]
return result
else:
return None

View file

@ -1,12 +1,13 @@
version:
runtime:
storage: /var/kiwi
shells:
- /bin/bash
env: null
markers:
project: .project
disabled: .disabled
version: 0.2.0
shells:
- /bin/bash
projects:
- name: admin
enabled: true
- test:
- test2: false
storage:
directory: /var/local/kiwi
network:
name: kiwi_hub
cidr: 10.22.46.0/24

View file

@ -0,0 +1,14 @@
from kiwi_scp.config import Config
import yaml
def main():
with open("./kiwi_scp/data/etc/kiwi_default.yml") as kc:
yml = yaml.safe_load(kc)
kiwi = Config(**yml)
print(repr(kiwi))
if __name__ == "__main__":
main()

View file

@ -7,6 +7,7 @@ authors = ["ldericher <40151420+ldericher@users.noreply.github.com>"]
[tool.poetry.dependencies]
python = "^3.6.1"
PyYAML = "^5.4.1"
pydantic = "^1.8.2"
[tool.poetry.dev-dependencies]
virtualenv = "^20.8.1"