##
import logging
-from pydantic import BaseModel, constr, PositiveInt
-from typing import Any, Dict, List
+from typing import Any, Dict, List, NoReturn
logger = logging.getLogger(__name__)
-class ConfigData(BaseModel):
- """Configuration data model."""
+def _validate_data(
+ config_data: Dict[str, Any], relation_data: Dict[str, Any]
+) -> NoReturn:
+ """Validate input data.
- openstack_default_granularity: PositiveInt
- global_request_timeout: PositiveInt
- log_level: constr(regex=r"^(INFO|DEBUG)$")
- collector_interval: PositiveInt
- evaluator_interval: PositiveInt
- database_commonkey: constr(min_length=1)
- vca_host: constr(min_length=1)
- vca_user: constr(min_length=1)
- vca_password: constr(min_length=1)
- vca_cacert: str
+ Args:
+ config_data (Dict[str, Any]): configuration data.
+ relation_data (Dict[str, Any]): relation data.
+ """
+ config_validators = {
+ "openstack_default_granularity": lambda value, _: isinstance(value, int)
+ and value > 0,
+ "global_request_timeout": lambda value, _: isinstance(value, int) and value > 0,
+ "log_level": lambda value, _: isinstance(value, str)
+ and value in ("INFO", "DEBUG"),
+ "collector_interval": lambda value, _: isinstance(value, int) and value > 0,
+ "evaluator_interval": lambda value, _: isinstance(value, int) and value > 0,
+ "database_commonkey": lambda value, _: isinstance(value, str)
+ and len(value) > 0,
+ "vca_host": lambda value, _: isinstance(value, str) and len(value) > 0,
+ "vca_user": lambda value, _: isinstance(value, str) and len(value) > 0,
+ "vca_password": lambda value, _: isinstance(value, str) and len(value) > 0,
+ "vca_cacert": lambda value, _: isinstance(value, str),
+ }
+ relation_validators = {
+ "message_host": lambda value, _: isinstance(value, str) and len(value) > 0,
+ "message_port": lambda value, _: isinstance(value, int) and value > 0,
+ "database_uri": lambda value, _: isinstance(value, str)
+ and value.startswith("mongodb://"),
+ "prometheus_host": lambda value, _: isinstance(value, str) and len(value) > 0,
+ "prometheus_port": lambda value, _: isinstance(value, int) and value > 0,
+ }
+ problems = []
+
+ for key, validator in config_validators.items():
+ valid = validator(config_data.get(key), config_data)
+
+ if not valid:
+ problems.append(key)
+ for key, validator in relation_validators.items():
+ valid = validator(relation_data.get(key), relation_data)
-class RelationData(BaseModel):
- """Relation data model."""
+ if not valid:
+ problems.append(key)
- message_host: constr(min_length=1)
- message_port: PositiveInt
- database_uri: constr(regex=r"^(mongodb://)")
- prometheus_host: constr(min_length=1)
- prometheus_port: PositiveInt
+ if len(problems) > 0:
+ raise ValueError("Errors found in: {}".format(", ".join(problems)))
def _make_pod_ports(port: int) -> List[Dict[str, Any]]:
if not image_info:
return None
- ConfigData(**(config))
- RelationData(**(relation_state))
+ _validate_data(config, relation_state)
ports = _make_pod_ports(port)
env_config = _make_pod_envconfig(config, relation_state)