Refactoring WIP

cosmo/clients/netbox_v4.py

@@ -251,24 +251,30 @@ def _fetch_data(self, kwargs):
               device_list(filters: {
                 name: { i_exact: $device },
               }) {
+                __typename
                 id
                 name
                 serial
 
                 device_type {
+                  __typename
                   slug
                 }
                 platform {
+                  __typename
                   manufacturer {
+                    __typename
                     slug
                   }
                   slug
                 }
                 primary_ip4 {
+                  __typename
                   address
                 }
 
                 interfaces {
+                  __typename
                   id
                   name
                   enabled
@@ -277,29 +283,36 @@ def _fetch_data(self, kwargs):
                   mtu
                   description
                   vrf {
+                    __typename
                     id
                   }
                   lag {
+                    __typename
                     id
                   }
                   ip_addresses {
+                    __typename
                     address
                   }
                   untagged_vlan {
+                    __typename
                     id
                     name
                     vid
                   }
                   tagged_vlans {
+                    __typename
                     id
                     name
                     vid
                   }
                   tags {
+                    __typename
                     name
                     slug
                   }
                   parent {
+                    __typename
                     id
                     mtu
                   }

cosmo/common.py

@@ -0,0 +1,8 @@
+def head(l):
+    return None if not l else l[0]
+
+
+def without_keys(d, keys) -> dict:
+    if type(keys) != list:
+        keys = [keys]
+    return {k: v for k,v in d.items() if k not in keys}

cosmo/types.py

@@ -0,0 +1,181 @@
+import abc
+from copy import deepcopy
+from .common import head, without_keys
+
+
+class AbstractNetboxType(abc.ABC, dict):
+    __parent = None
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.__mappings = {}
+        for c in AbstractNetboxType.__subclasses__():
+            self.__mappings.update(c.register())
+        for k, v in without_keys(self, "__parent").items():
+            self[k] = self.convert(v)
+
+    def convert(self, item):
+        if isinstance(item, dict):
+            if "__typename" in item.keys():
+                c = self.__mappings[item["__typename"]]
+                #            self descending in tree
+                return c({k: self.convert(v) for k, v in without_keys(item, "__parent").items()} | {"__parent": self})
+            else:
+                return item
+        elif isinstance(item, list):
+            replacement = []
+            for i in item:
+                #                  self descending in tree
+                replacement.append(self.convert(i))
+            return replacement
+        else:
+            return item
+
+    @classmethod
+    def _getNetboxType(cls):
+        # classes should have the same name as the type name
+        # if not, you can override in parent class
+        return cls.__name__
+
+    @classmethod
+    def register(cls) -> dict:
+        return {cls._getNetboxType(): cls}
+
+    def getParent(self):
+        return self['__parent']
+
+    def __deepcopy__(self, memo):
+        # I'm using convert because we have to rebuild the circular reference
+        # tree, since we cannot use the old references and thus __parent
+        # becomes invalid. Implementing __deepcopy__ is better than implementing
+        # workarounds for object instances in client code.
+        return self.__class__().convert(self.__class__(
+            {k: deepcopy(v,memo) for k, v in without_keys(self, "__parent").items()}
+        ))
+
+    def __repr__(self):
+        return self._getNetboxType()
+
+
+class AbstractManufacturerStrategy(abc.ABC):
+    def matches(self, manuf_slug):
+        return True if manuf_slug == self.mySlug() else False
+    @abc.abstractmethod
+    def mySlug(self):
+        pass
+    @abc.abstractmethod
+    def getRoutingInstanceName(self):
+        pass
+    @abc.abstractmethod
+    def getManagementInterfaceName(self):
+        pass
+    @abc.abstractmethod
+    def getBmcInterfaceName(self):
+        pass
+
+class JuniperManufacturerStrategy(AbstractManufacturerStrategy):
+    def mySlug(self):
+        return "juniper"
+    def getRoutingInstanceName(self):
+        return "mgmt_junos"
+    def getManagementInterfaceName(self):
+        return "fxp0"
+    def getBmcInterfaceName(self):
+        return None
+
+class RtBrickManufacturerStrategy(AbstractManufacturerStrategy):
+    def mySlug(self):
+        return "rtbrick"
+    def getRoutingInstanceName(self):
+        return "mgmt"
+    def getManagementInterfaceName(self):
+        return "ma1"
+    def getBmcInterfaceName(self):
+        return "bmc0"
+
+# POJO style store
+class DeviceType(AbstractNetboxType):
+    manufacturer_strategy: AbstractManufacturerStrategy = None
+
+    def getPlatformManufacturer(self):
+        return self.getPlatform().getManufacturer().getSlug()
+
+    # can't @cache, non-hashable
+    def getManufacturerStrategy(self):
+        if self.manufacturer_strategy:
+            return self.manufacturer_strategy
+        else:
+            slug = self.getPlatformManufacturer()
+            for c in AbstractManufacturerStrategy.__subclasses__():
+                if c().matches(slug):
+                    self.manufacturer_strategy = c(); break
+            return self.manufacturer_strategy
+
+    def getInterfaceByName(self, name):
+        l = list(
+            filter(
+                lambda i: i.getInterfaceName() == name,
+                self.getInterfaces()
+            )
+        )
+        return l if l != [] else None
+
+    def getRoutingInstance(self):
+        return self.getManufacturerStrategy().getRoutingInstanceName()
+
+    def getManagementInterface(self):
+        return head(self.getInterfaceByName(
+            self.getManufacturerStrategy().getManagementInterfaceName()
+        ))
+
+    def getBmcInterface(self):
+        return head(self.getInterfaceByName(
+            self.getManufacturerStrategy().getBmcInterfaceName()
+        ))
+
+    def getDeviceType(self):
+        return self['device_type']
+
+    def getPlatform(self):
+        return self['platform']
+
+    def getInterfaces(self):
+        return self['interfaces']
+
+
+class DeviceTypeType(AbstractNetboxType):
+    pass
+
+
+class PlatformType(AbstractNetboxType):
+    def getManufacturer(self):
+        return self['manufacturer']
+
+
+class ManufacturerType(AbstractNetboxType):
+    def getSlug(self):
+        return self['slug']
+
+
+class IPAddressType(AbstractNetboxType):
+    pass
+
+
+class InterfaceType(AbstractNetboxType):
+    def __repr__(self):
+        return super().__repr__() + f"({self.getInterfaceName()})"
+
+    def getInterfaceName(self):
+        return self['name']
+
+
+class VRFType(AbstractNetboxType):
+    pass
+
+
+class TagType(AbstractNetboxType):
+    pass
+
+
+class VLANType(AbstractNetboxType):
+    pass

cosmo/visitors.py

@@ -0,0 +1,49 @@
+import abc
+from copy import deepcopy
+from functools import singledispatchmethod
+
+
+class AbstractNoopNetboxTypesVisitor(abc.ABC):
+    def __init__(self, *args, **kwargs):
+        for c in AbstractNoopNetboxTypesVisitor.__subclasses__():
+            self.accept.register(
+                c,
+                lambda self, o: self._dictLikeTemplateMethod(o)
+            )
+
+    @singledispatchmethod
+    def accept(self, o):
+        raise NotImplementedError(f"unsupported type {o}")
+
+    @accept.register
+    def _(self, o: int):
+        return o
+
+    @accept.register
+    def _(self, o: None):
+        return o
+
+    @accept.register
+    def _(self, o: str):
+        return o
+
+    def _dictLikeTemplateMethod(self, o):
+        o = deepcopy(o)
+        keys = list(o.keys())
+        for key in keys:
+            self._mutateDictKVTemplateMethod(o, key)
+        return o
+
+    def _mutateDictKVTemplateMethod(self, o, key):
+        o[key] = self.accept(o[key])
+
+    @accept.register
+    def _(self, o: dict) -> dict:
+        return self._dictLikeTemplateMethod(o)
+
+    @accept.register
+    def _(self, o: list) -> list:
+        o = deepcopy(o)
+        for i, v in enumerate(o):
+            o[i] = self.accept(v)
+        return o