Loose transformation for pyrsistent. Similar to the transform method, but it works even if there are missing keys.

loose_transform.py

from typing import List, Callable, Union

from pyrsistent import PVector, PMap, ny, freeze


def loose_transform(structure: Union[PVector,PMap], path: List, command: Callable):
    if not path:
        return command(structure) if callable(command) else command
    key_spec = path[0]
    if not callable(key_spec):
        key_spec_original = key_spec
        key_spec = lambda ix: ix == key_spec_original
    if isinstance(structure, PVector):
        return freeze([
            loose_transform(item, path[1:], command) if key_spec(ix) else item
            for ix, item in enumerate(structure)
        ])
    return freeze({
        k: loose_transform(v, path[1:], command) if key_spec(k) else v
        for k, v in structure.items()
    })

test_loose_transform.py

import pyrsistent
from pyrsistent import v, m, freeze, ny
import pytest

from loose_transform import loose_transform

composer = m(birth=m(place="Salzburg", year=1777), name="Felix")
composer_upper = m(birth=m(place="SALZBURG", year=1777), name="Felix")
composer_const = m(birth=m(place="Linz", year=1777), name="Felix")
lyrics = m(name="John")
arranged_by = m(name="Jack", birth=m(year=1977))


@pytest.mark.parametrize(
    "command, known_result",
    [
        (
            str.upper,
            m(
                composer=composer_upper,
                lyrics=lyrics,
                arranged_by=arranged_by
            )
        ),
        (
            "Linz",
            m(
                composer=composer_const,
                lyrics=lyrics,
                arranged_by=arranged_by
            )
        ),
    ]
)
def test_missing_different_level(command, known_result):
    data = m(
        composer=composer,
        lyrics=lyrics,
        arranged_by=arranged_by
    )
    result = loose_transform(data, [ny, "birth", "place"], command)
    assert result == known_result


@pytest.mark.parametrize(
    "data, known_result",
    [
        (
            m(timeout="13", description="send email"),
            m(timeout=13, description="send email"),
        ),
        (
            m(description="send email"),
            m(description="send email"),
        ),
    ]
)
def test_convert_timeout_to_seconds_missing_timeout_one_level(data, known_result):
    result = loose_transform(data, ["timeout"], int)
    assert result == known_result


def test_convert_timeout_to_seconds_missing_timeout_two_levels():
    action_list = v(
        m(timeout="13", description="send email"),
        m(description="start database"),
    )
    new_action_list = [
        dict(timeout=13, description="send email"),
        dict(description="start database"),
    ]
    result = loose_transform(action_list, (ny, "timeout"), int)