Base of Record type registration

octodns/record/__init__.py

@@ -64,7 +64,11 @@ class Delete(Change):
         return f'Delete {self.existing}'
 
 
-class ValidationError(Exception):
+class RecordException(Exception):
+    pass
+
+
+class ValidationError(RecordException):
 
     @classmethod
     def build_message(cls, fqdn, reasons):
@@ -80,6 +84,18 @@ class ValidationError(Exception):
 class Record(EqualityTupleMixin):
     log = getLogger('Record')
 
+    _CLASSES = {}
+
+    @classmethod
+    def register_type(cls, _type, _class):
+        existing = cls._CLASSES.get(_type, None)
+        if existing:
+            module = existing.__module__
+            name = existing.__name__
+            msg = f'Type "{_type}" already registered by {module}.{name}'
+            raise RecordException(msg)
+        cls._CLASSES[_type] = _class
+
     @classmethod
     def new(cls, zone, name, data, source=None, lenient=False):
         name = str(name)
@@ -89,24 +105,7 @@ class Record(EqualityTupleMixin):
         except KeyError:
             raise Exception(f'Invalid record {fqdn}, missing type')
         try:
-            _class = {
-                'A': ARecord,
-                'AAAA': AaaaRecord,
-                'ALIAS': AliasRecord,
-                'CAA': CaaRecord,
-                'CNAME': CnameRecord,
-                'DNAME': DnameRecord,
-                'LOC': LocRecord,
-                'MX': MxRecord,
-                'NAPTR': NaptrRecord,
-                'NS': NsRecord,
-                'PTR': PtrRecord,
-                'SPF': SpfRecord,
-                'SRV': SrvRecord,
-                'SSHFP': SshfpRecord,
-                'TXT': TxtRecord,
-                'URLFWD': UrlfwdRecord,
-            }[_type]
+            _class = cls._CLASSES[_type]
         except KeyError:
             raise Exception(f'Unknown record type: "{_type}"')
         reasons = _class.validate(name, fqdn, data)
@@ -804,11 +803,17 @@ class ARecord(_DynamicMixin, _GeoMixin, Record):
     _value_type = Ipv4List
 
 
+Record.register_type('A', ARecord)
+
+
 class AaaaRecord(_DynamicMixin, _GeoMixin, Record):
     _type = 'AAAA'
     _value_type = Ipv6List
 
 
+Record.register_type('AAAA', AaaaRecord)
+
+
 class AliasValue(_TargetValue):
     pass
 
@@ -826,6 +831,9 @@ class AliasRecord(_ValueMixin, Record):
         return reasons
 
 
+Record.register_type('ALIAS', AliasRecord)
+
+
 class CaaValue(EqualityTupleMixin):
     # https://tools.ietf.org/html/rfc6844#page-5
 
@@ -877,6 +885,9 @@ class CaaRecord(_ValuesMixin, Record):
     _value_type = CaaValue
 
 
+Record.register_type('CAA', CaaRecord)
+
+
 class CnameRecord(_DynamicMixin, _ValueMixin, Record):
     _type = 'CNAME'
     _value_type = CnameValue
@@ -890,11 +901,17 @@ class CnameRecord(_DynamicMixin, _ValueMixin, Record):
         return reasons
 
 
+Record.register_type('CNAME', CnameRecord)
+
+
 class DnameRecord(_DynamicMixin, _ValueMixin, Record):
     _type = 'DNAME'
     _value_type = DnameValue
 
 
+Record.register_type('DNAME', DnameRecord)
+
+
 class LocValue(EqualityTupleMixin):
     # TODO: work out how to do defaults per RFC
 
@@ -1071,6 +1088,9 @@ class LocRecord(_ValuesMixin, Record):
     _value_type = LocValue
 
 
+Record.register_type('LOC', LocRecord)
+
+
 class MxValue(EqualityTupleMixin):
 
     @classmethod
@@ -1141,6 +1161,9 @@ class MxRecord(_ValuesMixin, Record):
     _value_type = MxValue
 
 
+Record.register_type('MX', MxRecord)
+
+
 class NaptrValue(EqualityTupleMixin):
     VALID_FLAGS = ('S', 'A', 'U', 'P')
 
@@ -1219,6 +1242,9 @@ class NaptrRecord(_ValuesMixin, Record):
     _value_type = NaptrValue
 
 
+Record.register_type('NAPTR', NaptrRecord)
+
+
 class _NsValue(object):
 
     @classmethod
@@ -1246,6 +1272,9 @@ class NsRecord(_ValuesMixin, Record):
     _value_type = _NsValue
 
 
+Record.register_type('NS', NsRecord)
+
+
 class PtrValue(_TargetValue):
 
     @classmethod
@@ -1279,6 +1308,9 @@ class PtrRecord(_ValuesMixin, Record):
         return self.values[0]
 
 
+Record.register_type('PTR', PtrRecord)
+
+
 class SshfpValue(EqualityTupleMixin):
     VALID_ALGORITHMS = (1, 2, 3, 4)
     VALID_FINGERPRINT_TYPES = (1, 2)
@@ -1343,6 +1375,9 @@ class SshfpRecord(_ValuesMixin, Record):
     _value_type = SshfpValue
 
 
+Record.register_type('SSHFP', SshfpRecord)
+
+
 class _ChunkedValuesMixin(_ValuesMixin):
     CHUNK_SIZE = 255
     _unescaped_semicolon_re = re.compile(r'\w;')
@@ -1392,6 +1427,9 @@ class SpfRecord(_ChunkedValuesMixin, Record):
     _value_type = _ChunkedValue
 
 
+Record.register_type('SPF', SpfRecord)
+
+
 class SrvValue(EqualityTupleMixin):
 
     @classmethod
@@ -1474,6 +1512,9 @@ class SrvRecord(_ValuesMixin, Record):
         return reasons
 
 
+Record.register_type('SRV', SrvRecord)
+
+
 class _TxtValue(_ChunkedValue):
     pass
 
@@ -1483,6 +1524,9 @@ class TxtRecord(_ChunkedValuesMixin, Record):
     _value_type = _TxtValue
 
 
+Record.register_type('TXT', TxtRecord)
+
+
 class UrlfwdValue(EqualityTupleMixin):
     VALID_CODES = (301, 302)
     VALID_MASKS = (0, 1, 2)
@@ -1558,3 +1602,6 @@ class UrlfwdValue(EqualityTupleMixin):
 class UrlfwdRecord(_ValuesMixin, Record):
     _type = 'URLFWD'
     _value_type = UrlfwdValue
+
+
+Record.register_type('URLFWD', UrlfwdRecord)

tests/test_octodns_record.py

@@ -10,9 +10,9 @@ from unittest import TestCase
 from octodns.record import ARecord, AaaaRecord, AliasRecord, CaaRecord, \
     CaaValue, CnameRecord, DnameRecord, Create, Delete, GeoValue, LocRecord, \
     LocValue, MxRecord, MxValue, NaptrRecord, NaptrValue, NsRecord, \
-    PtrRecord, Record, SshfpRecord, SshfpValue, SpfRecord, SrvRecord, \
-    SrvValue, TxtRecord, Update, UrlfwdRecord, UrlfwdValue, ValidationError, \
-    _Dynamic, _DynamicPool, _DynamicRule
+    PtrRecord, Record, RecordException, SshfpRecord, SshfpValue, SpfRecord, \
+    SrvRecord, SrvValue, TxtRecord, Update, UrlfwdRecord, UrlfwdValue, \
+    ValidationError, _Dynamic, _DynamicPool, _DynamicRule
 from octodns.zone import Zone
 
 from helpers import DynamicProvider, GeoProvider, SimpleProvider
@@ -21,6 +21,12 @@ from helpers import DynamicProvider, GeoProvider, SimpleProvider
 class TestRecord(TestCase):
     zone = Zone('unit.tests.', [])
 
+    def test_registration(self):
+        with self.assertRaises(RecordException) as ctx:
+            Record.register_type('A', None)
+        self.assertEqual('Type "A" already registered by '
+                         'octodns.record.ARecord', str(ctx.exception))
+
     def test_lowering(self):
         record = ARecord(self.zone, 'MiXeDcAsE', {
             'ttl': 30,