rougail-user-data-environment/src/rougail/user_data_environment/data.py

107 lines
4 KiB
Python

"""
Silique (https://www.silique.fr)
Copyright (C) 2024-2025
distribued with GPL-2 or later license
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""
import os
from rougail.tiramisu import CONVERT_OPTION
from rougail.config import RougailConfig
from rougail.error import ExtensionError
from tiramisu.error import ValueOptionError
class RougailUserDataEnvironment:
def __init__(
self,
config: "Config",
*,
rougailconfig: RougailConfig = None,
):
# this is the tiramisu config object
self.config = config
if rougailconfig is None:
rougailconfig = RougailConfig
user_data = rougailconfig["step.user_data"]
if "environment" not in user_data:
user_data.append("environment")
rougailconfig["step.user_data"] = user_data
user_data = rougailconfig["step.user_data"]
if "environment" not in user_data:
raise ExtensionError("environment is not set in step.user_data")
if "environment.with_secrets" in rougailconfig:
self.with_secrets = rougailconfig["environment.with_secrets"]
else:
self.with_secrets = True
self.custom_separator = rougailconfig["environment.custom_separator"]
if not rougailconfig["main_namespace"]:
self.default_environment_name = rougailconfig["environment.default_environment_name"]
self.errors = []
self.warnings = []
def run(self):
values = self.parse()
return [
{
"source": "environment variable",
"errors": self.errors,
"warnings": self.warnings,
"values": values,
"options": {
"multi_separator": ",",
"needs_convert": True,
"allow_secrets_variables": self.with_secrets,
},
}
]
def parse(self):
variables = {}
self.prefixes = []
for option in self.config:
if not option.isoptiondescription() or option.group_type() != "namespace":
break
self.prefixes.append(option.name().upper() + ".")
else:
return self.get_rougail_environment(0)
# no namespace then we filter the ROUGAIL_ environment variables
self.prefixes = [self.default_environment_name + "_"]
return self.get_rougail_environment(len(self.prefixes[0]))
def get_rougail_environment(self, len_env):
"""gets all the rougail environment variables and their values
:sample: {'VARINT': '5', 'VARNAME34': '58, 22', 'VARNAME2': 'tata',
'VARNAME1': 'titi', 'MYFAMILY.VARNAME3': 'spam'}
:returns: rougail environment variables as a key/value dict
"""
return {
envvar[len_env:].lower(): envval
for envvar, envval in self.get_correct_envs()
}
def get_correct_envs(self):
for envvar, envval in os.environ.items():
if self.custom_separator:
envvar = envvar.replace(self.custom_separator, '.')
for prefix in self.prefixes:
if envvar.startswith(prefix):
yield envvar, envval
for prefix in self.prefixes:
if envvar.startswith(prefix):
yield envvar, envval