Skip to content

alejandrogallo/konfigurazioa

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

14 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Introduction

This is a typed configuration, allowing for:

  • [X] Basic types
  • [X] Optional types
  • [X] Lists (typed or untyped)
  • [X] Union types
  • [ ] Callables (lambdas etc)
  • [ ] Python expressions
  • [ ] Guards
  • [ ] User defined types

Configuration file concept

A program will define some configuration settings. Every setting can be defined globally or within a namespace. For instance the following configuratoin

port: 80
base: index.html
proxy: None

my-namespace:
  port: 90
  base: main.html
  proxy: 121.12.41.2

defines port, base and proxy settings. However, these settings are overriden within the namespace my-namespace.

The defining configuration in the main program for this case would be

port:
  default: 80
  type: Int
  doc: |
    Port for the main application to listen to.
base:
  default: index.html
  type: String
  doc: Static file to serve in the root.
proxy:
  default: None
  type: Optional[String]
  doc: Proxy to reroute your main application.

Python configuration

A pythonic version of the configuration is also allowed

c.port = 80
c.base = "index.html"
c.proxy = None

with c.namespace("my-namespace") as p:
    p.port = 90
    p.base = "base.html"
    p.proxy = "121.12.41.2"

Implementation

The main objectives of this library should be:

  • [ ] correctly parsing the configuration file.
  • [ ] checking that the values of the settings check against the defined type schema provided by the library.
  • [ ] for the library user, be sure of getting the correct type of a value provided in the configuration, if an error occurs, throw well-defined and well-documented exceptions.

Types

The main structure for the types is given by a class with a match and a parse function.

from typing import ( Any, List, TypeVar, Generic, Union, Optional
                   , Callable, Tuple, NamedTuple
                   )
import re
import os

Ty = NamedTuple("Ty", [ ("name", str)
                      , ("match", Callable[[Any], bool])
                      , ("parse", Callable[[Any], Any])])


def ty_matcher(t: Ty, v:Any) -> bool:
    try:
        t.parse(v)
    except ValueError:
        return False
    else:
        return True

Now we can define the basic python types as wrapped types.

def make_basic_wrapper(t: type, name: str) -> Ty:
    tt = Ty(name=name,
            match=lambda x: ty_matcher(tt, x),
            parse=lambda x: t(x))  # type: Ty
    return tt


def make_bool() -> Ty:
    def parse_bool(t: Ty, v: Any) -> bool:

        if isinstance(v, bool):
            return v
        else:

            if v in ["true", "True"]:
                return True

            if v in ["false", "False"]:
                return False
        raise ValueError("Invalid value for type {} ({})"
                         .format(t.name, v))

    tt = Ty(name="Bool",
            match=lambda x: ty_matcher(tt, x),
            parse=lambda x: parse_bool(tt, x))
    return tt


Int = make_basic_wrapper(int, "Int")
Float = make_basic_wrapper(float, "Float")
String = make_basic_wrapper(str, "String")
Bool = make_bool()
PythonExpression \
    = Ty("PythonExpression",
         match=lambda x: True,
         parse=lambda x: eval(x))
PythonExpressionWithEnvironment \
    = Ty("PythonExpressionWithEnvironment",
         match=lambda x: True,
         parse=lambda x: eval(x, {"env": os.environ}))


def make_optional(t: Ty) -> Ty:
    tt = Ty(name="Optional[{}]".format(t.name),
            match=lambda x: ty_matcher(tt, x),
            parse=lambda x: None if (x in [None, "None"]) else t.parse(x))  # type: Ty
    return tt


def make_list(t: Ty) -> Ty:
    def parse_list(_t: Ty, v: Any) -> List[Any]:
        if isinstance(v, list):
            _list = v
        else:
            _list = re.findall(r"[^,\[\]()]+", str(v))
            if not _list:
                raise SyntaxError("Invalid list: '{}'".format(v))
        return [_t.parse(e) for e in _list]
    tt = Ty(name="List[{}]".format(t.name),
            match=lambda x: ty_matcher(tt, x),
            parse = lambda x: parse_list(t, x))
    return tt


def make_union(t: Ty, s: Ty) -> Ty:
    def parse_union(tt: Ty, _t: Ty, _s: Ty, x: Any) -> Any:
        wrap_types = (_t, _s)
        for i in range(2):
            try:
                t = wrap_types[i]
                return t.parse(x)
            except ValueError:
                pass
        raise ValueError("Invalid value for type {} ({})"
                         .format(tt.name, x))
    tt = Ty(name="Union[{},{}]".format(t.name, s.name),
            match=lambda x: ty_matcher(tt, x),
            parse = lambda x: parse_union(tt, t, s, x))
    return tt


def string_to_union(name: str) -> Optional[Ty]:
    m = re.match(r"Union\[([^\[\]]+)\s*,\s*([^\[\]]+)\s*\]", name)
    if not m:
        return None
    fst = string_to_type(m.group(1))
    snd = string_to_type(m.group(2))
    return make_union(fst, snd)


def string_to_list(name: str) -> Optional[Ty]:
    m = re.match(r"List\[([^\[\]]+)\]", name)
    if not m:
        return None
    t = string_to_type(m.group(1))
    return make_list(t)


def string_to_optional(name: str) -> Optional[Ty]:
    m = re.match(r"Optional\[([^\[\]]+)\]", name)
    if not m:
        return None
    t = string_to_type(m.group(1))
    return make_optional(t)


TYPES = [ lambda x: Int if re.match(Int.name, x) else None
        , lambda x: Float if re.match(Float.name, x) else None
        , lambda x: String if re.match(String.name, x) else None
        , lambda x: Bool if re.match(Bool.name, x) else None
        , string_to_optional
        , string_to_list
        , string_to_union
        , lambda x: PythonExpressionWithEnvironment
                    if re.match(PythonExpressionWithEnvironment.name, x)
                    else None
        , lambda x: PythonExpression
                    if re.match(PythonExpression.name, x)
                    else None
        ]  # List[Callable[[str], Optional[Ty]]]


def string_to_type(name: str, types: List[Callable[[str], Optional[Ty]]]=TYPES) -> Ty:
    for t in types:
        _t = t(name)
        if _t:
            return _t
    raise TypeError("Type {} not recognised".format(name))

Tests

Configuration file

The configuration consists of a Schema written in yaml and a user configuration written in some suitable configuration language like toml, yaml etc…

Schema

from typing import NamedTuple, Any, List, Callable, Dict
import konfigurazioa.types as kt
import yaml


Guard = NamedTuple("Guard", [ ("message", str)
                            , ("callable", Callable[[Any], bool])
                            ])


SchemaAtom = NamedTuple( "SchemaAtom"
                       , [ ("name", str)
                         , ("type", kt.Ty)
                         , ("doc", str)
                         # The type will be checked at parsing time
                         , ("default", Any)
                         , ("guards", List[Guard])
                         ]
                       )


Schema = List[SchemaAtom]


def guard_from_dict(d: Dict[str, str]) -> Guard:
    _l = eval(d["callable"])
    assert callable(_l), "Guard's callable must be a callable object"
    return Guard(d["message"], _l)


def schema_from_file(filepath: str) -> Schema:
    schema = []  # type: Schema
    with open(filepath) as f:
        raw_schema = yaml.load(f, Loader=yaml.FullLoader)
    for key in raw_schema:
        string_default = raw_schema[key]["default"]
        string_type = raw_schema[key]["type"]
        t = kt.string_to_type(string_type)
        default = t.parse(string_default)
        guards = raw_schema[key].get("guards", [])
        schema.append(SchemaAtom( name=key
                                , type=t
                                , doc=raw_schema[key]["doc"]
                                , default=default
                                , guards=[guard_from_dict(g) for g in guards]))
    return schema

Tests

Configuration

What should be a good API for reading in a user configuration?

import yaml
from typing import Dict, Any, NamedTuple, Optional, TypeVar
from collections import defaultdict

from konfigurazioa.schema import Schema, SchemaAtom
import konfigurazioa.types as kt

DataAtom = NamedTuple("DataAtom", [ ("value", Any)
                                  , ("type", kt.Ty)
                                  , ("name", str)
                                  ])
SectionData = Dict[str, DataAtom]
ConfigData = Dict[Optional[str], SectionData]


def validate_data(val: Any, s: SchemaAtom) -> DataAtom:
    v = s.type.parse(val)
    # Run guards
    for guard in s.guards:
        if not guard.callable(v):
            raise ValueError("Incorrect value for '{s}' ({v}): {m}"
                             .format(s=s.name, v=v, m=guard.message))
    return DataAtom(value=v,
                    type=s.type,
                    name=s.name)


def dict_to_section_data(data: Dict[str, Any],
                         schema: Schema,
                         section: str) -> SectionData:
    result = {}  # type: SectionData
    for key, val in data.items():
        _s = [s for s in schema if s.name == key]
        if not _s:
            raise ValueError("Key {} is not a valid setting name".format(key))
        s = _s[0]
        result[s.name] = validate_data(val, s)
    return result


def default_data(schema: Schema) -> SectionData:
    return {
        s.name: DataAtom(value=s.default,
                         type=s.type,
                         name=s.name)
        for s in schema
    }


def parse_data_from_schema(data: Dict[str, Any], schema: Schema) -> ConfigData:
    result = defaultdict(lambda: default_data(schema))  # type: ConfigData
    for key, val in data.items():
        _s = [s for s in schema if s.name == key]
        if not _s and not isinstance(val, dict):
            raise ValueError("Key {} is not a valid setting name".format(key))
        elif not _s and isinstance(val, dict):
            section = key
            result[section].update(dict_to_section_data(val, schema, section))
        else:
            s = _s[0]
            result[None][key] = validate_data(val, s)
    return result


class Configuration:

    def __init__(self, filepath: str, schema: Schema) -> None:
        self.__filepath = filepath  # type: str
        self.__data = {}  # type: ConfigData
        self.__schema = schema  # type: Schema
        self.__read()

    def __read(self) -> None:
        with open(self.__filepath) as f:
            data = yaml.load(f, Loader=yaml.FullLoader)
        self.__data = parse_data_from_schema(data, self.__schema)

    def update_from_file(self, path: str) -> None:
        c = Configuration(path, self.__schema)
        self.__data.update(c.__data)

    def get(self, key: str, section: Optional[str]=None) -> Any:
        return self.__data[section][key].value

Sphinx documentation

import docutils
from docutils.parsers.rst import Directive
from typing import Any, List

import konfigurazioa.schema as ks


SETTING_TEMPLATE = """\
.. _config-{name}:
**{name}** (config-{name}_)
    - type: {type}
    - default: {default}

"""


class Setting(Directive):  # type: ignore

    has_content = True
    optional_arguments = 2
    required_arguments = 1
    #option_spec = dict(schema=str, description=str)
    add_index = True

    def run(self) -> Any:
        name = self.arguments[0]
        schema_path = self.options.get('schema')
        schema = ks.schema_from_file(schema_path)
        _s = [s for s in schema if s.name == name]
        if not _s:
            raise ValueError("{} not in schema".format(name))
        s = _s[0]
        default = s.default
        source = self.state_machine.input_lines.source(
            self.lineno - self.state_machine.input_offset - 1)

        default_list = []

        if '\n' in str(default):
            default_list.append("        .. code::")
            default_list.append("")
            for lindef in default.split('\n'):
                default_list.append(3*"    " + lindef)
        else:
            default_list.append(" ``{value}``"
                                .format(value=default))

        lines = SETTING_TEMPLATE.format(default="\n".join(default_list),
                                        type=s.type.name,
                                        name=name).split("\n")

        newViewList = docutils.statemachine.ViewList(lines)
        self.content = newViewList + self.content # type: List[str]

        node = docutils.nodes.paragraph()
        node.document = self.state.document
        self.state.nested_parse(self.content, self.content_offset, node)
        return node.children


def setup(app: Any) -> None:
    app.add_directive('konfigurazioa-setting', Setting)