Module static_topo_impl.model.stackstate
Expand source code
from datetime import datetime
from typing import Any, Dict, List
import pytz
from schematics import Model
from schematics.transforms import blacklist, wholelist
from schematics.types import (BaseType, DictType, ListType, ModelType,
StringType)
from schematics.types import TimestampType as DefaultTimestampType
from schematics.types import URLType
class AnyType(BaseType):
def __init__(self, **kwargs):
super(AnyType, self).__init__(**kwargs)
class TimestampType(DefaultTimestampType):
def to_primitive(self, value, context=None):
value.astimezone(pytz.UTC)
return int(round(value.timestamp())) # seconds
class ComponentType(Model):
name: str = StringType(required=True)
class Options:
roles = {"public": wholelist()}
class Relation(Model):
external_id: str = StringType(required=True, serialized_name="externalId")
relation_type: ComponentType = ModelType(ComponentType, serialized_name="type")
source_id: str = StringType(required=True, serialized_name="sourceId")
target_id: str = StringType(required=True, serialized_name="targetId")
properties: Dict[str, Any] = DictType(AnyType(), required=True, default={"labels": []}, serialized_name="data")
class Options:
roles = {"public": wholelist()}
def set_type(self, name: str):
if self.relation_type is None:
self.relation_type = ComponentType({"name": name})
else:
self.relation_type.name = name
def get_type(self) -> str:
if self.relation_type is None:
return ""
else:
return self.relation_type.name
class ComponentProperties(Model):
name: str = StringType(required=True)
layer: str = StringType(default="Unknown")
domain: str = StringType(default="Unknown")
environment: str = StringType(default="Unknown")
labels: List[str] = ListType(StringType(), default=[])
identifiers: List[str] = ListType(StringType(), default=[])
custom_properties: Dict[str, Any] = DictType(AnyType(), default={})
class Options:
roles = {"public": wholelist()}
def add_label(self, label: str):
self.labels.append(label)
def add_label_kv(self, key: str, value: str):
self.labels.append(f"{key}:{value}")
def add_identifier(self, identifier: str):
if identifier not in self.identifiers:
self.identifiers.append(identifier)
def update_properties(self, properties: Dict[str, Any]):
self.custom_properties.update(properties)
def add_property(self, name: str, value: Any):
self.custom_properties[name] = value
def get_property(self, name: str):
return self.custom_properties[name]
def dedup_labels(self):
labels = set(self.labels)
self.labels = list(labels)
class Component(Model):
uid: str = StringType(required=True, serialized_name="externalId")
component_type: ComponentType = ModelType(ComponentType, serialized_name="type")
properties: ComponentProperties = ModelType(
ComponentProperties, required=True, default=ComponentProperties(), serialized_name="data"
)
relations: List[Relation] = ListType(ModelType(Relation), default=[])
class Options:
roles = {"public": blacklist("relations")}
def set_type(self, name: str):
if self.component_type is None:
self.component_type = ComponentType({"name": name})
else:
self.component_type.name = name
def get_type(self) -> str:
return self.component_type.name
def get_name(self) -> str:
return self.properties.name
def set_name(self, name: str):
self.properties.name = name
class HealthCheckState(Model):
check_id: str = StringType(required=True, serialized_name="checkStateId")
check_name: str = StringType(required=True, serialized_name="name")
topo_identifier: str = StringType(required=True, serialized_name="topologyElementIdentifier")
message: str = StringType(required=False)
health: str = StringType(required=True, choices=["CLEAR", "DEVIATING", "CRITICAL"])
class Options:
roles = {"public": wholelist()}
class SourceLink(Model):
title = StringType(required=True)
url = URLType(fqdn=False, required=True)
class Options:
roles = {"public": wholelist()}
class EventContext(Model):
category: str = StringType(required=True, choices=["Activities", "Alerts", "Anomalies", "Changes", "Others"])
data: Dict[str, Any] = DictType(AnyType, default={})
element_identifiers: List[str] = ListType(StringType, required=False, default=[])
source: str = StringType(required=True, default="StaticTopology")
source_links: List[SourceLink] = ListType(ModelType(SourceLink), required=False, default=[])
class Options:
roles = {"public": wholelist()}
serialize_when_none = False
class Event(Model):
context: EventContext = ModelType(EventContext, required=True, default=EventContext())
event_type: str = StringType(required=True)
msg_title: str = StringType(required=True)
msg_text: str = StringType(required=True)
source_type_name: str = StringType(required=True, default="StaticTopology")
tags: List[str] = ListType(StringType, required=False, default=[])
timestamp: datetime = TimestampType(required=True)
class Options:
roles = {"public": wholelist()}
Classes
class AnyType (**kwargs)-
A base class for Types in a Schematics model. Instances of this class may be added to subclasses of
Modelto define a model schema.Validators that need to access variables on the instance can be defined be implementing methods whose names start with
validate_and accept one parameter (in addition toself):param required: Invalidate field when value is None or is not supplied. Default: False. :param default: When no data is provided default to this value. May be a callable. Default: None. :param serialized_name: The name of this field defaults to the class attribute used in the model. However if the field has another name in foreign data set this argument. Serialized data will use this value for the key name too. :param deserialize_from: A name or list of named fields for which foreign data sets are searched to provide a value for the given field. This only effects inbound data. :param choices: A list of valid choices. This is the last step of the validator chain. :param validators: A list of callables. Each callable receives the value after it has been converted into a rich python type. Default: [] :param serialize_when_none: Dictates if the field should appear in the serialized data even if the value is None. Default: None. :param messages: Override the error messages with a dict. You can also do this by subclassing the Type and defining a
MESSAGESdict attribute on the class. A metaclass will merge all theMESSAGESand override the resulting dict with instance levelmessagesand assign toself.messages. :param metadata: Dictionary for storing custom metadata associated with the field. To encourage compatibility with external tools, we suggest these keys for common metadata: - label : Brief human-readable label - description : Explanation of the purpose of the field. Used for help, tooltips, documentation, etc. - example : A concrete example of the field usage.Expand source code
class AnyType(BaseType): def __init__(self, **kwargs): super(AnyType, self).__init__(**kwargs)Ancestors
- schematics.types.base.BaseType
Class variables
var MESSAGES
class Component (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class Component(Model): uid: str = StringType(required=True, serialized_name="externalId") component_type: ComponentType = ModelType(ComponentType, serialized_name="type") properties: ComponentProperties = ModelType( ComponentProperties, required=True, default=ComponentProperties(), serialized_name="data" ) relations: List[Relation] = ListType(ModelType(Relation), default=[]) class Options: roles = {"public": blacklist("relations")} def set_type(self, name: str): if self.component_type is None: self.component_type = ComponentType({"name": name}) else: self.component_type.name = name def get_type(self) -> str: return self.component_type.name def get_name(self) -> str: return self.properties.name def set_name(self, name: str): self.properties.name = nameAncestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar component_type : ComponentTypevar properties : ComponentPropertiesvar relations : List[Relation]var uid : str
Methods
def get_name(self) ‑> str-
Expand source code
def get_name(self) -> str: return self.properties.name def get_type(self) ‑> str-
Expand source code
def get_type(self) -> str: return self.component_type.name def set_name(self, name: str)-
Expand source code
def set_name(self, name: str): self.properties.name = name def set_type(self, name: str)-
Expand source code
def set_type(self, name: str): if self.component_type is None: self.component_type = ComponentType({"name": name}) else: self.component_type.name = name
- schematics.deprecated.patch_models.
class ComponentProperties (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class ComponentProperties(Model): name: str = StringType(required=True) layer: str = StringType(default="Unknown") domain: str = StringType(default="Unknown") environment: str = StringType(default="Unknown") labels: List[str] = ListType(StringType(), default=[]) identifiers: List[str] = ListType(StringType(), default=[]) custom_properties: Dict[str, Any] = DictType(AnyType(), default={}) class Options: roles = {"public": wholelist()} def add_label(self, label: str): self.labels.append(label) def add_label_kv(self, key: str, value: str): self.labels.append(f"{key}:{value}") def add_identifier(self, identifier: str): if identifier not in self.identifiers: self.identifiers.append(identifier) def update_properties(self, properties: Dict[str, Any]): self.custom_properties.update(properties) def add_property(self, name: str, value: Any): self.custom_properties[name] = value def get_property(self, name: str): return self.custom_properties[name] def dedup_labels(self): labels = set(self.labels) self.labels = list(labels)Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar custom_properties : Dict[str, Any]var domain : strvar environment : strvar identifiers : List[str]var labels : List[str]var layer : strvar name : str
Methods
def add_identifier(self, identifier: str)-
Expand source code
def add_identifier(self, identifier: str): if identifier not in self.identifiers: self.identifiers.append(identifier) def add_label(self, label: str)-
Expand source code
def add_label(self, label: str): self.labels.append(label) def add_label_kv(self, key: str, value: str)-
Expand source code
def add_label_kv(self, key: str, value: str): self.labels.append(f"{key}:{value}") def add_property(self, name: str, value: Any)-
Expand source code
def add_property(self, name: str, value: Any): self.custom_properties[name] = value def dedup_labels(self)-
Expand source code
def dedup_labels(self): labels = set(self.labels) self.labels = list(labels) def get_property(self, name: str)-
Expand source code
def get_property(self, name: str): return self.custom_properties[name] def update_properties(self, properties: Dict[str, Any])-
Expand source code
def update_properties(self, properties: Dict[str, Any]): self.custom_properties.update(properties)
- schematics.deprecated.patch_models.
class ComponentType (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class ComponentType(Model): name: str = StringType(required=True) class Options: roles = {"public": wholelist()}Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar name : str
- schematics.deprecated.patch_models.
class Event (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class Event(Model): context: EventContext = ModelType(EventContext, required=True, default=EventContext()) event_type: str = StringType(required=True) msg_title: str = StringType(required=True) msg_text: str = StringType(required=True) source_type_name: str = StringType(required=True, default="StaticTopology") tags: List[str] = ListType(StringType, required=False, default=[]) timestamp: datetime = TimestampType(required=True) class Options: roles = {"public": wholelist()}Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar context : EventContextvar event_type : strvar msg_text : strvar msg_title : strvar source_type_name : strvar timestamp : datetime.datetime
- schematics.deprecated.patch_models.
class EventContext (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class EventContext(Model): category: str = StringType(required=True, choices=["Activities", "Alerts", "Anomalies", "Changes", "Others"]) data: Dict[str, Any] = DictType(AnyType, default={}) element_identifiers: List[str] = ListType(StringType, required=False, default=[]) source: str = StringType(required=True, default="StaticTopology") source_links: List[SourceLink] = ListType(ModelType(SourceLink), required=False, default=[]) class Options: roles = {"public": wholelist()} serialize_when_none = FalseAncestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar category : strvar data : Dict[str, Any]var element_identifiers : List[str]var source : strvar source_links : List[SourceLink]
- schematics.deprecated.patch_models.
class HealthCheckState (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class HealthCheckState(Model): check_id: str = StringType(required=True, serialized_name="checkStateId") check_name: str = StringType(required=True, serialized_name="name") topo_identifier: str = StringType(required=True, serialized_name="topologyElementIdentifier") message: str = StringType(required=False) health: str = StringType(required=True, choices=["CLEAR", "DEVIATING", "CRITICAL"]) class Options: roles = {"public": wholelist()}Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar check_id : strvar check_name : strvar health : strvar message : strvar topo_identifier : str
- schematics.deprecated.patch_models.
class Relation (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class Relation(Model): external_id: str = StringType(required=True, serialized_name="externalId") relation_type: ComponentType = ModelType(ComponentType, serialized_name="type") source_id: str = StringType(required=True, serialized_name="sourceId") target_id: str = StringType(required=True, serialized_name="targetId") properties: Dict[str, Any] = DictType(AnyType(), required=True, default={"labels": []}, serialized_name="data") class Options: roles = {"public": wholelist()} def set_type(self, name: str): if self.relation_type is None: self.relation_type = ComponentType({"name": name}) else: self.relation_type.name = name def get_type(self) -> str: if self.relation_type is None: return "" else: return self.relation_type.nameAncestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar external_id : strvar properties : Dict[str, Any]var relation_type : ComponentTypevar source_id : strvar target_id : str
Methods
def get_type(self) ‑> str-
Expand source code
def get_type(self) -> str: if self.relation_type is None: return "" else: return self.relation_type.name def set_type(self, name: str)-
Expand source code
def set_type(self, name: str): if self.relation_type is None: self.relation_type = ComponentType({"name": name}) else: self.relation_type.name = name
- schematics.deprecated.patch_models.
class SourceLink (raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)-
Enclosure for fields and validation. Same pattern deployed by Django models, SQLAlchemy declarative extension and other developer friendly libraries.
:param Mapping raw_data: The data to be imported into the model instance. :param Mapping deserialize_mapping: Can be used to provide alternative input names for fields. Values may be strings or lists of strings, keyed by the actual field name. :param bool partial: Allow partial data to validate. Essentially drops the
required=Truesettings from field definitions. Default: True :param bool strict: Complain about unrecognized keys. Default: TrueExpand source code
class SourceLink(Model): title = StringType(required=True) url = URLType(fqdn=False, required=True) class Options: roles = {"public": wholelist()}Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Optionsvar titlevar url
- schematics.deprecated.patch_models.
class TimestampType (formats=None, parser=None, drop_tzinfo=False, **kwargs)-
A variant of
DateTimeTypethat exports itself as a Unix timestamp instead of an ISO 8601 string. Always setstzd='require'andconvert_tz=True.Expand source code
class TimestampType(DefaultTimestampType): def to_primitive(self, value, context=None): value.astimezone(pytz.UTC) return int(round(value.timestamp())) # secondsAncestors
- schematics.types.base.TimestampType
- schematics.types.base.DateTimeType
- schematics.types.base.BaseType
Class variables
var MESSAGES
Methods
def to_primitive(self, value, context=None)-
Convert internal data to a value safe to serialize.
Expand source code
def to_primitive(self, value, context=None): value.astimezone(pytz.UTC) return int(round(value.timestamp())) # seconds