import inspect
import logging
from inspect import Parameter, signature, getsourcefile
from typing import TypedDict, NotRequired, Unpack, Optional
from q2_sdk.core.cli.textui import colored, indent, puts
from q2_sdk.core.configuration import settings
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.tools import utils
[docs]
class InstallStepArguments(TypedDict):
allow_editable: NotRequired[bool]
required_install_step_attrs: NotRequired[list[str]]
[docs]
class InstallStep:
"""Base class for DbPlan Steps"""
def __init__(self, **kwargs: Unpack[InstallStepArguments]):
"""
:param allow_editable: If False, will force all InstallStepAttributes' to is_editable==False
:param required_install_step_attrs: An optional list of strings for names of install steps that will be forced to is_required==True
"""
self.logger = logging.getLogger(__name__)
self.hq_credentials = HqCredentials(
settings.HQ_CREDENTIALS.hq_url,
settings.HQ_CREDENTIALS.csr_user,
settings.HQ_CREDENTIALS.csr_pwd,
settings.HQ_CREDENTIALS.aba,
)
self.extension_name: Optional[str] = None
self.install_order = 10
self.allow_editable = kwargs.get("allow_editable", True)
self.not_install_var_list = []
self.is_internal = "q2_sdk/core/install_steps" in getsourcefile(self.__class__)
self._required_install_step_attrs = kwargs.get(
"required_install_step_attrs", []
)
self.is_central = kwargs.get("is_central", False)
def print_install_vars(self):
for k, val in self.install_vars.items():
color = "yellow" if k not in self.editable_attrs else "cyan"
with indent(4):
if len(str(val.value)) < 150:
puts(getattr(colored, color)("{}: '{}'".format(k, val.value)))
else:
puts(
getattr(colored, color)(
"{}: {} of length {}".format(
k, type(val.value), len(str(val.value))
)
)
)
@property
def install_vars(self):
"""Returns all attributes that are also InstallStepAttributes"""
vars_to_return = {}
for key, value in vars(self).items():
if isinstance(value, InstallStepAttribute):
vars_to_return[key] = value
if key in self._required_install_step_attrs:
value.is_required = True
raw_params = signature(self.__init__).parameters
params = [
x
for x in raw_params
if raw_params[x].kind == Parameter.POSITIONAL_OR_KEYWORD
]
non_install_vars = [
x
for x in params
if x not in vars_to_return and x not in self.not_install_var_list
]
for var in non_install_vars:
default = raw_params[var].default
default = default if default != inspect._empty else None
vars_to_return[var] = InstallStepAttribute(
default,
is_editable=True,
is_bool=default in (True, False),
is_required=var in self._required_install_step_attrs,
)
return vars_to_return
@property
def required_attrs(self) -> list[str]:
"""Returns all install steps with the .is_required flag set to True"""
return [a for a in self.install_vars if self.install_vars[a].is_required]
@property
def editable_attrs(self) -> list[str]:
"""Returns all attributes with the .is_editable flag set to True"""
return [a for a in self.install_vars if self.install_vars[a].is_editable]
@property
def has_editable_attrs(self) -> bool:
"""True if at least one attribute has .is_editable == True"""
return bool(len(self.editable_attrs)) and self.allow_editable
@property
def hidden_attrs(self) -> list[str]:
"""Returns all attributes with the .is_hidden flag set to True"""
return [a for a in self.install_vars if self.install_vars[a].is_hidden]
[docs]
async def install(self):
"""Must be overridden in inherited class to work with install"""
for step_attr_name in self.required_attrs:
step_attr = getattr(self, step_attr_name)
if step_attr.value == "":
raise ValueError(f"Required value ({step_attr_name}) cannot be empty")
[docs]
async def uninstall(self):
"""Must be overridden in inherited class to work with remove_form"""
return
def __json__(self):
return self.install_vars
def __repr__(self):
return str(self.install_vars)
def __eq__(self, other):
if isinstance(other, InstallStep):
other = other.install_vars
return self.install_vars == other
class InstallStepAttribute:
def __init__(
self,
value,
is_editable=False,
is_hidden=False,
use_textarea=False,
is_bool=False,
is_required=False,
):
"""
:param is_required: Makes sure that the value is not an empty string
"""
self.__value = None
self.default = value
self.is_editable = is_editable
self.is_hidden = is_hidden
self.use_textarea = use_textarea
if isinstance(value, bool):
is_bool = True
self.is_bool = is_bool
self.is_required = is_required
self.value = value
def __json__(self):
return {
"value": self.value,
"default": self.default,
"is_editable": self.is_editable,
"is_hidden": self.is_hidden,
"use_textarea": self.use_textarea,
"is_bool": self.is_bool,
"is_required": self.is_required,
}
def __repr__(self):
return str(vars(self))
def __eq__(self, other):
return vars(self) == vars(other)
@property
def value(self):
return self.__value
@value.setter
def value(self, value):
if self.is_bool:
self.__value = utils.to_bool(value)
return
self.__value = value