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
Model
to 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
MESSAGES
dict attribute on the class. A metaclass will merge all theMESSAGES
and override the resulting dict with instance levelmessages
and 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=True
settings 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 = name
Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Options
var component_type : ComponentType
var properties : ComponentProperties
var 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=True
settings 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 Options
var custom_properties : Dict[str, Any]
var domain : str
var environment : str
var identifiers : List[str]
var labels : List[str]
var layer : str
var 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=True
settings 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 Options
var 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=True
settings 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 Options
var context : EventContext
var event_type : str
var msg_text : str
var msg_title : str
var source_type_name : str
var 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=True
settings 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 = False
Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Options
var category : str
var data : Dict[str, Any]
var element_identifiers : List[str]
var source : str
var 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=True
settings 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 Options
var check_id : str
var check_name : str
var health : str
var message : str
var 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=True
settings 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.name
Ancestors
- schematics.deprecated.patch_models.
.Model - schematics.deprecated.ModelCompatibilityMixin
- schematics.models.Model
Class variables
var Options
var external_id : str
var properties : Dict[str, Any]
var relation_type : ComponentType
var source_id : str
var 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=True
settings 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 Options
var title
var url
- schematics.deprecated.patch_models.
class TimestampType (formats=None, parser=None, drop_tzinfo=False, **kwargs)
-
A variant of
DateTimeType
that 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())) # seconds
Ancestors
- 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