Add context to Templating, cleanly handle records w/o template method, doc

octodns/processor/templating.py

@@ -6,9 +6,58 @@ from octodns.processor.base import BaseProcessor
 
 
 class Templating(BaseProcessor):
+    '''
+    Record templating using python format. For simple records like TXT and CAA
+    that is the value itself. For multi-field records like MX or SRV it's the
+    text portions, exchange and target respectively.
 
-    def __init__(self, id, *args, **kwargs):
+    Example Processor Config:
+
+        templating:
+          class: octodns.processor.templating.Templating
+          # Any k/v present in context will be passed into the .format method and
+          # thus be available as additional variables in the template. This is all
+          # optional.
+          context:
+            key: value
+            another: 42
+
+    Example Records:
+
+        foo:
+          type: TXT
+          value: The zone this record lives in is {zone_name}. There are {zone_num_records} record(s).
+
+        bar:
+          type: MX
+          values:
+            - preference: 1
+              exchange: mx1.{zone_name}.mail.mx.
+            - preference: 1
+              exchange: mx2.{zone_name}.mail.mx.
+
+    Note that validations for some types reject values with {}. When
+    encountered the best option is to use record level `lenient: true`
+    https://github.com/octodns/octodns/blob/main/docs/records.md#lenience
+
+    Note that if you need to add dynamic context you can create a custom
+    processor that inherits from Templating and passes them into the call to
+    super, e.g.
+
+        class MyTemplating(Templating):
+            def __init__(self, *args, context={}, **kwargs):
+                context['year'] = lambda desired, sources: datetime.now().strftime('%Y')
+                super().__init__(*args, context, **kwargs)
+
+    See https://docs.python.org/3/library/string.html#custom-string-formatting
+    for details on formatting options. Anything possible in an `f-string` or
+    `.format` should work here.
+
+    '''
+
+    def __init__(self, id, *args, context={}, **kwargs):
         super().__init__(id, *args, **kwargs)
+        self.context = context
 
     def process_source_zone(self, desired, sources):
         sources = sources or []
@@ -18,6 +67,13 @@ class Templating(BaseProcessor):
             'zone_encoded_name': desired.name,
             'zone_num_records': len(desired.records),
             'zone_source_ids': ', '.join(s.id for s in sources),
+            # add any extra context provided to us, if the value is a callable
+            # object call it passing our params so that arbitrary dynamic
+            # context can be added for use in formatting
+            **{
+                k: (v(desired, sources) if callable(v) else v)
+                for k, v in self.context.items()
+            },
         }
 
         def params(record):
@@ -36,12 +92,18 @@ class Templating(BaseProcessor):
 
         for record in desired.records:
             if hasattr(record, 'values'):
+                if record.values and not hasattr(record.values[0], 'template'):
+                    # the (custom) value type does not support templating
+                    continue
                 new_values = [v.template(params(record)) for v in record.values]
                 if record.values != new_values:
                     new = record.copy()
                     new.values = new_values
                     desired.add_record(new, replace=True)
             else:
+                if not hasattr(record.value, 'template'):
+                    # the (custom) value type does not support templating
+                    continue
                 new_value = record.value.template(params(record))
                 if record.value != new_value:
                     new = record.copy()

tests/test_octodns_processor_templating.py

@@ -6,10 +6,59 @@ from unittest import TestCase
 from unittest.mock import call, patch
 
 from octodns.processor.templating import Templating
-from octodns.record import Record
+from octodns.record import Record, ValueMixin, ValuesMixin
 from octodns.zone import Zone
 
 
+class DummySource:
+
+    def __init__(self, id):
+        self.id = str(id)
+
+
+class CustomValue(str):
+
+    @classmethod
+    def validate(cls, *args, **kwargs):
+        return []
+
+    @classmethod
+    def process(cls, v):
+        if isinstance(v, (list, tuple)):
+            return (CustomValue(i) for i in v)
+        return CustomValue(v)
+
+    @classmethod
+    def parse_rdata_text(cls, *args, **kwargs):
+        pass
+
+    def __init__(self, *args, **kwargs):
+        self._asked_for = set()
+
+    def rdata_text(self):
+        pass
+
+    def __getattr__(self, item):
+        self._asked_for.add(item)
+        raise AttributeError('nope')
+
+
+class Single(ValueMixin, Record):
+    _type = 'S'
+    _value_type = CustomValue
+
+
+Record.register_type(Single, 'S')
+
+
+class Multiple(ValuesMixin, Record):
+    _type = 'M'
+    _value_type = CustomValue
+
+
+Record.register_type(Multiple, 'M')
+
+
 def _find(zone, name):
     return next(r for r in zone.records if r.name == name)
 
@@ -75,14 +124,29 @@ class TemplatingTest(TestCase):
         noop = _find(got, 'noop')
         self.assertEqual('Nothing to template here.', noop.values[0])
 
-    @patch('octodns.record.TxtValue.template')
-    def test_params(self, mock_template):
+    def test_no_template(self):
         templ = Templating('test')
 
-        class DummySource:
+        zone = Zone('unit.tests.', [])
+        s = Record.new(zone, 's', {'type': 'S', 'ttl': 42, 'value': 'string'})
+        zone.add_record(s)
 
-            def __init__(self, id):
-                self.id = id
+        m = Record.new(
+            zone, 'm', {'type': 'M', 'ttl': 43, 'values': ('string', 'another')}
+        )
+        zone.add_record(m)
+
+        # this should check for the template method on our values that don't
+        # have one
+        templ.process_source_zone(zone, None)
+        # and these should make sure that the value types were asked if they
+        # have a template method
+        self.assertEqual({'template'}, s.value._asked_for)
+        self.assertEqual({'template'}, m.values[0]._asked_for)
+
+    @patch('octodns.record.TxtValue.template')
+    def test_params(self, mock_template):
+        templ = Templating('test')
 
         zone = Zone('unit.tests.', [])
         record_source = DummySource('record')
@@ -123,3 +187,41 @@ class TemplatingTest(TestCase):
             ),
             mock_template.call_args,
         )
+
+    def test_context(self):
+        templ = Templating(
+            'test',
+            context={
+                # static
+                'the_answer': 42,
+                # dynamic
+                'the_date': lambda _, __: 'today',
+                # uses a param
+                'num_sources': lambda z, ss: len(ss),
+            },
+        )
+
+        zone = Zone('unit.tests.', [])
+        txt = Record.new(
+            zone,
+            'txt',
+            {
+                'type': 'TXT',
+                'ttl': 42,
+                'values': (
+                    'the_answer: {the_answer}',
+                    'the_date: {the_date}',
+                    'num_sources: {num_sources}',
+                ),
+            },
+        )
+        zone.add_record(txt)
+
+        got = templ.process_source_zone(
+            zone, tuple(DummySource(i) for i in range(3))
+        )
+        txt = _find(got, 'txt')
+        self.assertEqual(3, len(txt.values))
+        self.assertEqual('num_sources: 3', txt.values[0])
+        self.assertEqual('the_answer: 42', txt.values[1])
+        self.assertEqual('the_date: today', txt.values[2])