2.7.2 - add Caster

CHANGELOG.md

 # v2.7.2
 
 * changed behaviour of MergingSource a bit
+* added Caster to configuration schema
 
 # v2.7.1
 

docs/configuration/schema.rst

@@ -21,6 +21,8 @@ you should instantiate a Descriptor. Descriptor reflects how your config is nest
 
 .. autoclass:: satella.configuration.schema.Dict
 
+.. autoclass:: satella.configuration.schema.Caster
+
 Then there is a descriptor that makes it possible for a value to have one of two types:
 
 .. autoclass:: satella.configuration.schema.Union
@@ -73,6 +75,7 @@ Available string types are:
 * **any** - Descriptor
 * **bool** - Boolean
 * **union** - Union
+* **caster** - Caster
 
 Lists you define as following
 
@@ -103,3 +106,16 @@ Dicts are more simple. Each key contains the key that should be present in the d
 You load it using the following function:
 
 .. autofunction:: satella.configuration.schema.descriptor_from_dict
+
+Casters you define as
+
+::
+
+    {
+        "type": "caster"
+        "cast_to": "name of a built-in or a fully qualified class ID"
+    }
+
+If cast_to is not a builtin, it specifies a full path to a class,
+which will be loaded using
+:func:`satella.imports.import_class`

satella/__init__.py

-__version__ = '2.7.2_a2'
+__version__ = '2.7.2'

satella/configuration/schema.py

@@ -3,6 +3,8 @@ import re
 import typing as tp
 
 from satella.coding.concurrent.callablegroup import CallableGroup
+from satella.coding.recast_exceptions import rethrow_as
+from satella.imports import import_class
 from ..exceptions import ConfigurationValidationError, ConfigurationSchemaError
 
 __all__ = [
@@ -14,6 +16,7 @@ __all__ = [
     'create_key',
     'must_be_type',
     'must_be_one_of',
+    'Caster',
     'CheckerCondition',
     'ConfigDictValue',
     'descriptor_from_dict',
@@ -219,6 +222,25 @@ def create_key(descriptor: Descriptor, name: str, optional: bool = False,
     return descriptor
 
 
+class Caster(Descriptor):
+    """
+    A value must be ran through a function.
+
+    Use like:
+
+    >>> class Environment(enum.IntEnum):
+    >>>     PRODUCTION = 0
+    >>> assert Caster(Environment)(0) == Environment.PRODUCTION
+    """
+
+    def __init__(self, to_cast: tp.Callable[[tp.Any], tp.Any]):
+        self.to_cast = to_cast
+
+    @rethrow_as(ValueError, ConfigurationValidationError)
+    def __call__(self, value: ConfigDictValue):
+        return self.to_cast(value)
+
+
 class Dict(Descriptor):
     """
     This entry must be a dict, having at least specified keys.
@@ -304,7 +326,8 @@ class Union(Descriptor):
 
 
 BASE_LOOKUP_TABLE = {'int': Integer, 'float': Float, 'str': String, 'ipv4': IPv4, 'list': List,
-                     'dict': Dict, 'any': Descriptor, 'bool': Boolean, 'union': Union}
+                     'dict': Dict, 'any': Descriptor, 'bool': Boolean, 'union': Union,
+                     'caster': Caster}
 
 
 def _get_descriptor_for(key: str, value: tp.Any) -> Descriptor:
@@ -324,6 +347,14 @@ def _get_descriptor_for(key: str, value: tp.Any) -> Descriptor:
             if type_ == 'list':
                 of = _get_descriptor_for('', value.get('of', ''))
                 args = (of,)
+            elif type_ == 'caster':
+                cast_to = value.get('cast_to')
+                dict_to_look_in = globals()
+                dict_to_look_in.update(locals())
+                if cast_to in dict_to_look_in:
+                    args = (dict_to_look_in[cast_to],)
+                else:
+                    args = (import_class(cast_to),)
             elif type_ == 'union':
                 args = [_get_descriptor_for('', x) for x in value.get('of', [])]
             optional, default = False, None

tests/test_configuration/test_schema.py

+import enum
 import os
 import tempfile
 import unittest
@@ -7,8 +8,28 @@ from satella.configuration.sources import DirectorySource
 from satella.exceptions import ConfigurationValidationError
 
 
+class Environment(enum.IntEnum):
+    PRODUCTION = 0
+
+
 class TestSchema(unittest.TestCase):
 
+    def test_caster(self):
+        ps = Caster(Environment)
+        self.assertEqual(ps(0), Environment.PRODUCTION)
+
+    def test_descriptor_from_schema_caster(self):
+        schema = {
+            "key": {
+                'type': 'caster',
+                'cast_to': 'tests.test_configuration.test_schema.Environment'
+            }
+        }
+
+        s = descriptor_from_dict(schema)
+
+        self.assertEqual(s({'key': 0}), {'key': Environment.PRODUCTION})
+
     def test_union(self):
         ps = Union(List(), Dict(keys=[create_key(String(), 'a')]))