"""Deployment configuration entities."""
import copy
import re
# Permit identifiers up to 96 bytes in length
_DEPLOY_ID_RE = re.compile("^[A-Za-z0-9\\-]{1,96}$")
# Supported deployment kinds
DEPLOYMENT_KINDS = {"helm"} # Use helm controller process
[docs]
class Deployment:
"""Deployment entity.
Collects configuration information relating to a cluster
configuration change.
"""
def __init__(self, dpl_id, kind, args):
"""
Create a new deployment structure.
:param dpl_id: Deployment ID
:param kind: Kind of deployment (method by which it is applied)
:param args: Kind-specific deployment arguments
:returns: Deployment object
"""
# Get parameter dictionary
self._dict = {
"dpl_id": str(dpl_id),
"kind": str(kind),
"args": dict(copy.deepcopy(args)),
}
# Validate
if kind not in DEPLOYMENT_KINDS:
raise ValueError(f"Unknown deployment kind {kind}!")
if not _DEPLOY_ID_RE.match(self.dpl_id):
raise ValueError(f"Deployment ID {self.dpl_id} not permissible!")
[docs]
def to_dict(self):
"""Return data as dictionary."""
return self._dict
@property
def dpl_id(self):
"""Return the deployment id."""
return self._dict["dpl_id"]
@property
def kind(self):
"""Return deployment kind."""
return self._dict.get("kind")
@property
def args(self):
"""Return deployment arguments."""
return self._dict["args"]
def __repr__(self):
"""Produce object representation."""
content = ",".join([f"{k}={v!r}" for k, v in self.to_dict().items()])
return f"entity.Deployment({content})"
def __eq__(self, other):
"""Equality check."""
return self.to_dict() == other.to_dict()