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 to self)

: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 the MESSAGES and override the resulting dict with instance level messages and assign to self.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: True

Expand 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_typeComponentType
var propertiesComponentProperties
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
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: True

Expand 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)
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: True

Expand 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
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: True

Expand 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 contextEventContext
var event_type : str
var msg_text : str
var msg_title : str
var source_type_name : str
var tags : List[str]
var timestamp : datetime.datetime
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: True

Expand 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
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: True

Expand 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
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: True

Expand 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_typeComponentType
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

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: True

Expand 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
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 sets tzd='require' and convert_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