Settings deserialization for python: pydantic vs attrs+cattrs

config.dev.toml

secret_key = "3EEeX_v1exDXdE79CbLVDpQB4F1dmxm2cOC0eQB7Q1k="

[hops.aliyun]
ip = "112.112.112.2"
username = "kingo"

[hops.pi]
ip = "10.10.100.2"
username = "pi"

[hops.bridge]
ip = "172.10.100.100"
username = "kingo"

[tunnels.main]
hops = ["aliyun", "pi", "bridge"]

[forwards.mqtt]
tunnel = "main"
dest_host = "112.112.112.14"
dest_port = 1883
local_port = 1883

demo_cattrs.py

from base64 import urlsafe_b64decode
from ipaddress import ip_address, IPv4Address, IPv6Address
from typing import Optional, Dict, Union, List, AnyStr

import attr
import cattr
import toml


def not_empty(instance, attribute, value):
    if not value:
        raise TypeError(f'{attribute.name} must not be empty.')


def check_secret_key(instance, attribute, value):
    assert len(urlsafe_b64decode(value)) == 32, \
        f'"{attribute.name}" should be 32 url safe b64 encoded chars'


def force_bytes(s: AnyStr, encoding='utf-8') -> bytes:
    if isinstance(s, bytes):
        return s
    if isinstance(s, memoryview):
        return bytes(s)
    return str(s).encode(encoding)


@attr.define
class Hop:
    ip: Union[IPv4Address, IPv6Address]
    port: Optional[int] = None
    username: Optional[str] = None
    password: Optional[str] = None

    @property
    def connect_params(self) -> dict:
        rv = {'host': str(self.ip)}
        for field in ['port', 'username', 'password']:
            value = getattr(self, field, None)
            if value is not None:
                rv[field] = value
        return rv


@attr.define
class Tunnel:
    hops: List[str] = attr.field(factory=list, validator=not_empty)


@attr.define
class Settings:
    """ Demo settings. """

    secret_key: bytes = attr.ib(validator=check_secret_key, converter=force_bytes)
    hops: Dict[str, Hop]
    tunnels: Dict[str, Tunnel]

    def __attrs_post_init__(self):
        hop_names = set(self.hops.keys())
        for name, tunnel in self.tunnels.items():
            missing = set(tunnel.hops) - hop_names
            if missing:
                missing_names = ', '.join(map(str, missing))
                raise ValueError(
                    f'Cannot find hop[{missing_names}] required by tunnel[{name}]'
                )



if __name__ == '__main__':
    # With cattrs>=1.8.0, we can use Converter(prefer_attrib_converters=True) instead.
    converter = cattr.Converter()
    converter.register_structure_hook(bytes, lambda d, t: force_bytes(d))
    converter.register_structure_hook(
        Union[IPv4Address, IPv6Address],
        lambda d, t: ip_address(d),
    )

    with open('config.dev.toml') as fp:
        data = toml.load(fp)
        settings = converter.structure(data, Settings)
    print(settings)
# Settings(secret_key=b'3EEeX_v1exDXdE79CbLVDpQB4F1dmxm2cOC0eQB7Q1k=', hops={'aliyun': Hop(ip=IPv4Address('112.112.112.2'), port=None, username='kingo', password=None), 'pi': Hop(ip=IPv4Address('10.10.100.2'), port=None, username='pi', password=None), 'bridge': Hop(ip=IPv4Address('172.10.100.100'), port=None, username='kingo', password=None)}, tunnels={'main': Tunnel(hops=['aliyun', 'pi', 'bridge'])})

demo_pydantic.py

from base64 import urlsafe_b64decode
from ipaddress import IPv4Address, IPv6Address
from typing import Optional, Dict, Union, List

import pydantic
import toml


class Hop(pydantic.BaseModel):
    ip: Union[IPv4Address, IPv6Address]
    port: Optional[int]
    username: Optional[str]
    password: Optional[str]

    @property
    def connect_params(self):
        rv = {'host': str(self.ip)}
        for field in ['port', 'username', 'password']:
            value = getattr(self, field, None)
            if value is not None:
                rv[field] = value
        return rv


class Tunnel(pydantic.BaseModel):
    hops: List[str]


class Settings(pydantic.BaseSettings):
    """ Demo settings. """

    secret_key: pydantic.conbytes(min_length=32)
    hops: Dict[str, Hop] = pydantic.Field()
    tunnels: Dict[str, Tunnel] = pydantic.Field()

    class Config:
        env_prefix = 'SSH_FORWARD_'
        extra = 'ignore'

    @pydantic.validator('secret_key')  # noqa
    @classmethod
    def check_secret_key(cls, value: bytes) -> bytes:
        assert len(urlsafe_b64decode(value)) == 32, \
            'secret key should be 32 url safe b64 encoded chars'
        return value

    @pydantic.root_validator()  # noqa
    @classmethod
    def check_hops(cls, values):
        hops: Dict[str, Hop] = values['hops']
        tunnels: Dict[str, Tunnel] = values['tunnels']
        hop_names = set(hops.keys())
        for name, tunnel in tunnels.items():
            missing = set(tunnel.hops) - hop_names
            if missing:
                missing_names = ', '.join(map(str, missing))
                raise ValueError(
                    f'Cannot find hop[{missing_names}] required by tunnel[{name}]'
                )
        return values


if __name__ == '__main__':
    with open('config.dev.toml') as fp:
        data = toml.load(fp)
        settings = Settings(**data)
    print(settings)
# Settings(secret_key=b'3EEeX_v1exDXdE79CbLVDpQB4F1dmxm2cOC0eQB7Q1k=', hops={'aliyun': Hop(ip=IPv4Address('112.112.112.2'), port=None, username='kingo', password=None), 'pi': Hop(ip=IPv4Address('10.10.100.2'), port=None, username='pi', password=None), 'bridge': Hop(ip=IPv4Address('172.10.100.100'), port=None, username='kingo', password=None)}, tunnels={'main': Tunnel(hops=['aliyun', 'pi', 'bridge'])})