Integrations

Predicates are plain callables, so they work with any library that accepts a function returning a bool. For some libraries py-predicate provides ready-made wrappers.

annotated-types

annotated-types provides Predicate, which takes any callable returning a bool. Predicates can be used directly, no wrapper needed. annotated-types itself doesn’t validate anything; a library like pydantic enforces the constraint:

from typing import Annotated

from annotated_types import Predicate
from pydantic import BaseModel

from predicate import ge_lt_p, is_str_p

class Person(BaseModel):
    name: Annotated[str, Predicate(is_str_p)]
    age: Annotated[int, Predicate(ge_lt_p(0, 150))]

Person(name="alice", age=30)  # ok
Person(name="alice", age=200)  # raises ValidationError

beartype

beartype accepts any callable as a validator through beartype.vale.Is. Since predicates are callables, they can be used directly in type hints, no wrapper needed:

from typing import Annotated

from beartype import beartype
from beartype.vale import Is

from predicate import ge_le_p, is_list_of_p, is_str_p

Percentage = Annotated[int, Is[ge_le_p(0, 100)]]


@beartype
def set_volume(level: Percentage) -> Percentage:
    return level


@beartype
def join(words: Annotated[list, Is[is_list_of_p(is_str_p)]]) -> str:
    return " ".join(words)


set_volume(50)  # ok
set_volume(101)  # raises BeartypeCallHintParamViolation
join(["a", 1])  # raises BeartypeCallHintParamViolation

pandera

pandera accepts any callable as a check through pandera.Check. With element_wise=True the check is called per value, so predicates can be used directly, no wrapper needed:

import pandas as pd
import pandera.pandas as pa

from predicate import ge_lt_p, is_str_p

schema = pa.DataFrameSchema(
    {
        "name": pa.Column(checks=pa.Check(is_str_p, element_wise=True)),
        "age": pa.Column(int, pa.Check(ge_lt_p(0, 150), element_wise=True)),
    }
)

schema.validate(pd.DataFrame({"name": ["alice"], "age": [30]}))  # ok
schema.validate(pd.DataFrame({"name": ["alice"], "age": [200]}))  # raises SchemaError

ReactiveX

The predicate.rx module wraps the ReactiveX operators that take a predicate, so they can be used directly in a pipe. Install the optional dependency first:

pip install py_predicate[rx]

Then filter a stream with a predicate:

import reactivex as rx

from predicate import gt_p
from predicate.rx import filter_p

rx.of(1, 2, 3, 4).pipe(filter_p(gt_p(2))).subscribe(print)  # prints 3 and 4

The following operators are available:

Operator	Description
`filter_p`	only lets through items satisfying the predicate
`take_while_p`	lets through items as long as the predicate holds (`inclusive` keeps the first item that fails)
`skip_while_p`	skips items as long as the predicate holds
`partition_p`	splits a stream into `[matching, non_matching]` observables
`all_p`	emits whether all items satisfy the predicate
`some_p`	emits whether any item satisfies the predicate
`count_p`	emits the number of items satisfying the predicate
`first_p`	emits the first item satisfying the predicate, errors if there is none
`last_p`	emits the last item satisfying the predicate, errors if there is none
`single_p`	emits the only item satisfying the predicate, errors if there is not exactly one
`first_or_default_p`	like `first_p`, but emits `default_value` if there is none
`last_or_default_p`	like `last_p`, but emits `default_value` if there is none
`single_or_default_p`	like `single_p`, but emits `default_value` if there is none
`find_p`	emits the first item satisfying the predicate, or `None` if there is none
`find_index_p`	emits the index of the first item satisfying the predicate, or `-1` if there is none

all_p and count_p share their names with core predicates, so alias them when you need both:

from predicate import all_p
from predicate.rx import all_p as rx_all_p

Note

partition_p subscribes to the source once per output stream. With a cold source such as rx.of, the first subscription consumes all items before the second one starts. Use a hot source when both streams need to see the items, e.g. a Subject or a published source:

import reactivex as rx
import reactivex.operators as ops

from predicate import gt_p
from predicate.rx import partition_p

source = rx.of(1, 2, 3, 4).pipe(ops.publish())
matching, non_matching = source.pipe(partition_p(gt_p(2)))

matching.subscribe(print)  # prints 3 and 4
non_matching.subscribe(print)  # prints 1 and 2
source.connect()