Conditions and Pipes

Warning

Please, make sure you've covered Reference / Basics first.

Conditions

There are 2 conversions which allow to define conditions:

1. c.if_(condition, if_true, if_false) for a single condition
2. c.if_multiple(*condition_to_value_pairs, else_) obviously for multiple conditions

convtoolsdebug stdout

from convtools import conversion as c

# No. 1
converter = (
    c.iter(c.if_(c.this < 0, c.this * 2, c.this / 2))
    .as_type(list)
    .gen_converter(debug=True)
)
assert converter([-1, 0, 1]) == [-2, 0, 0.5]

# No. 2
converter = (
    c.iter(
        c.if_multiple(
            (c.this < 0, c.this * 2),
            (c.this == 0, 100),
            (c.this < 10, c.this / 2),
            else_=c.this / 10,
        )
    )
    .as_type(list)
    .gen_converter(debug=True)
)
assert converter([-1, 0, 1, 20]) == [-2, 100, 0.5, 2]

def converter(data_):
    try:
        return [(((i * 2) if (i < 0) else (i / 2))) for i in data_]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

def if_multiple_(data_):
    if data_ < 0:
        return data_ * 2
    if data_ == 0:
        return 100
    if data_ < 10:
        return data_ / 2
    return data_ / 10

def converter(data_):
    try:
        return [if_multiple_(i) for i in data_]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

Check and raise

There's a convenient helper, which checks whether a condition is met and returns input as is or raises c.ExpectException otherwise:

1. c.expect(condition, error_msg=None)
2. also as a method: c.attr("amount").expect(c.this < 10, "too big")

convtoolsdebug stdout

from convtools import conversion as c

# expect doesn't change input
converter = (
    c.iter(c.expect(c.this < 3, "too big") ** 10)
    .as_type(list)
    .gen_converter(debug=True)
)
assert converter(range(3)) == [0, 1, 1024]

# error_msg can be conversion itself
converter = (
    c.item("a")
    .expect(
        condition=c.this.len() > 3,
        error_msg=c.call_func("{} is too short".format, c.this),
    )
    .gen_converter(debug=True)
)
try:
    converter({"a": "val"})
except c.ExpectException as e:
    assert str(e) == "val is too short"

def expect(data_):
    if data_ < 3:
        return data_
    raise ExpectException("too big")

def converter(data_):
    try:
        return [(expect(i) ** 10) for i in data_]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

def expect(data_, *, __format=__naive_values__["__format"]):
    if len(data_) > 3:
        return data_
    raise ExpectException(__format(data_))

def converter(data_):
    try:
        return expect(data_["a"])
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

Pipes

Pipe is the most important conversion which allows to pass results of one conversion to another. The syntax is simple: first.pipe(second).

convtoolsdebug stdout

from convtools import conversion as c

converter = (
    c.iter(
        (c.item("value") + 1).pipe(
            c.if_(c.this < 0, c.this * c.this, c.this * 2)
        )
    )
    .as_type(list)
    .gen_converter(debug=True)
)
assert converter([{"value": -4}, {"value": 2}]) == [9, 6]

def pipe_(input_):
    return (input_ * input_) if (input_ < 0) else (input_ * 2)

def converter(data_):
    try:
        return [pipe_((i["value"] + 1)) for i in data_]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

You can also pipe to callables, have a look at its signature: pipe(next_conversion, *args, label_input=None, label_output=None, **kwargs). It accepts *args and **kwargs, which are passed to a callable after pipe input:

convtoolsdebug stdout

from datetime import datetime
from convtools import conversion as c

assert c.this.pipe(int).execute("123", debug=True) == 123

converter = (
    c.item("dt").pipe(datetime.strptime, "%m/%d/%Y").gen_converter(debug=True)
)
assert converter({"dt": "12/25/2000"}) == datetime(2000, 12, 25)

def converter(data_):
    try:
        return int(data_)
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

def converter(data_, *, __strptime=__naive_values__["__strptime"], __v=__naive_values__["__v"]):
    try:
        return __strptime(data_["dt"], __v)
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

and_then

and_then(conversion, condition=bool) method applies provided conversion if condition is true (by default condition is standard python's truth check). condition accepts both conversions and callables.

convtoolsdebug stdout

from convtools import conversion as c

# DEFAULT CONDITION
converter = (
    c.iter(c.this.and_then(c.this.as_type(int)))
    .as_type(list)
    .gen_converter(debug=True)
)
assert converter(["1", None, 2.0]) == [1, None, 2]

# CUSTOM CONDITION
converter = (
    c.iter(c.this.and_then(c.this + 10, condition=c.this != 1))
    .as_type(list)
    .gen_converter(debug=True)
)
assert converter(range(3)) == [10, 1, 12]

def converter(data_):
    try:
        return [(i and int(i)) for i in data_]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

def converter(data_):
    try:
        return [(((i + 10) if (i != 1) else i)) for i in data_]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

Labels

Warning

Despite the fact that convtools encourages a functional approach and working with immutable data, sometimes it's inevitable to use global variables. Anyway avoid using labels if possible.

There are two ways to label data for further use:

1. pipe method accepts label_input (applies to pipe's input data) and label_output (applies to the end result) keyword arguments, each of them is either:
   • str - label name
   • dict - label names to conversion map. Labels are put on results of conversions.
2. add_label - shortcut to pipe(This, label_input=label_name)

To reference previously labeled data use c.label("label_name").

convtoolsdebug stdout

from datetime import datetime
from convtools import conversion as c

converter = (
    c.this.add_label({"a": c.item("a")})
    .item("b")
    .iter({"a": c.label("a"), "b": c.this})
    .as_type(list)
    .gen_converter(debug=True)
)
# SAME
converter_2 = (
    c.this.pipe(c.item("b"), label_input={"a": c.item("a")})
    .iter({"a": c.label("a"), "b": c.this})
    .as_type(list)
    .gen_converter(debug=True)
)
input_data = {
    "a": 1,
    "b": [2, 3, 4],
}
expected_output = [{"a": 1, "b": 2}, {"a": 1, "b": 3}, {"a": 1, "b": 4}]
assert (
    converter(input_data) == expected_output
    and converter_2(input_data) == expected_output
)


# BETTER WITHOUT LABELS (HERE IT'S POSSIBLE)
converter_3 = (
    c.zip(a=c.repeat(c.item("a")), b=c.item("b"))
    .as_type(list)
    .gen_converter(debug=True)
)
assert converter_3(input_data) == expected_output

def pipe_(_labels, input_):
    _labels["a"] = input_["a"]
    return input_

def converter(data_):
    _labels = {}
    try:
        return [{"a": _labels["a"], "b": i} for i in pipe_(_labels, data_)["b"]]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

def pipe_(_labels, input_):
    _labels["a"] = input_["a"]
    return [{"a": _labels["a"], "b": i} for i in input_["b"]]

def converter(data_):
    _labels = {}
    try:
        return pipe_(_labels, data_)
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

def converter(data_, *, __repeat=__naive_values__["__repeat"]):
    try:
        return [{"a": i[0], "b": i[1]} for i in zip(__repeat(data_["a"]), data_["b"])]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise

Dispatch

Experimental feature

It was added on Feb 7, 2024 and may be stabilized ~ in half a year.

There are performance critical cases where it's desired to replace c.if_ and c.if_multiple with dict lookups. However it limits what can be used as keys as these need to be hashable.

Interface: c.this.dispatch(key, key_to_conv, default)

1. key defines a conversion, which gets a key
2. key_to_conv is a dict which maps keys to conversions
3. default is an optional default conversion, when the dict doesn't contain the key

convtoolsdebug stdout

from convtools import conversion as c


input_data = [
    {"version": "v1", "field1": 10},
    {"version": "v2", "field2": 20},
    {"version": "v3", "field": 30},
]

converter = (
    c.iter(
        c.this.dispatch(
            c.item("version"),
            {
                "v1": c.item("field1"),
                "v2": c.item("field2"),
            },
            default=c.item("field"),
        )
    )
    .as_type(list)
    .gen_converter(debug=True)
)

assert converter(input_data) == [10, 20, 30]

def branch(data_):
    return data_["field1"]

def branch_i(data_):
    return data_["field2"]

def branch_else(data_):
    return data_["field"]

def converter(data_, *, __v=__naive_values__["__v"], __branch_else=__naive_values__["__branch_else"]):
    try:
        return [__v.get(i["version"], __branch_else)(i) for i in data_]
    except __exceptions_to_dump_sources:
        __convtools__code_storage.dump_sources()
        raise