koalify 1.4.0

A compact predicate DSL for matching criteria against any object

koalify

A compact predicate DSL for matching criteria against any Python object. Zero runtime dependencies.

Installation

pip install koalify

Quick Start

from koalify import F, all_of, any_of

# Build rules with Python operators
is_eligible = (
    (F.status == "active")
    & (F.age >= 18)
    & (F.role.in_("admin", "moderator", "editor"))
    & F.score.between(50, 100)
)

# Evaluate against any object with attributes
is_eligible(user)  # True / False

# Nested fields
lives_in_london = F.address.city == "London"

# Compose with OR / NOT
can_access = is_eligible | (lives_in_london & ~(F.status == "banned"))

# Dynamic composition from a list
conditions = [F.status == "active", F.age >= 18]
rule = all_of(*conditions)

Examples

Dataclasses

from dataclasses import dataclass
from koalify import F, all_of

@dataclass
class Order:
    product: str
    quantity: int
    price: float
    fulfilled: bool

needs_review = (F.quantity > 100) & (F.price >= 500) & (F.fulfilled == False)

order = Order(product="Widget", quantity=200, price=750.0, fulfilled=False)
needs_review(order)  # True

Pydantic

from pydantic import BaseModel
from koalify import F, any_of

class Address(BaseModel):
    city: str
    country: str

class Customer(BaseModel):
    name: str
    tier: str
    address: Address

is_priority = (F.tier.in_("gold", "platinum")) | (F.address.country == "US")

customer = Customer(name="Alice", tier="gold", address=Address(city="London", country="UK"))
is_priority(customer)  # True

Item access (lists, dicts)

from dataclasses import dataclass
from koalify import F

@dataclass
class Event:
    name: str
    tags: list[str]
    metadata: dict[str, str]

event = Event(name="deploy", tags=["prod", "urgent"], metadata={"region": "eu-west-1"})

# List indexing
(F.tags[0] == "prod")(event)  # True

# Dict key access
(F.metadata["region"] == "eu-west-1")(event)  # True

# Mix with attribute access and composition
is_urgent_prod = (F.tags[0] == "prod") & (F.tags[1] == "urgent")
is_urgent_prod(event)  # True

Dynamic rule composition

from koalify import F, all_of

def build_filter(min_age: int | None = None, status: str | None = None, roles: set[str] | None = None):
    criteria = []
    if min_age is not None:
        criteria.append(F.age >= min_age)
    if status is not None:
        criteria.append(F.status == status)
    if roles is not None:
        criteria.append(F.role.in_(*roles))
    return all_of(*criteria) if criteria else lambda _: True

user_filter = build_filter(min_age=18, roles={"admin", "editor"})

Implication (AND-of-implications)

implies(antecedent, consequent) encodes "if A then B" — vacuously true when the antecedent does not hold. Combine with all_of(...) to express a table of conditional constraints, where each row contributes a clause that only "bites" when its selector matches.

from koalify import F, all_of, implies

# Per-(plan, kind) upload size limit. plan and kind select which row applies;
# size_mb is the value being constrained.
cases = [
    {"plan": "free", "kind": "image", "size_mb": 5},
    {"plan": "pro",  "kind": "video", "size_mb": 500},
]

within_limits = all_of(*[
    implies(
        (F.plan == c["plan"]) & (F.kind == c["kind"]),
        F.size_mb < c["size_mb"],
    )
    for c in cases
])

Uploads whose (plan, kind) matches no row pass vacuously. Swap all_of for any_of(... & ...) (DNF) if you want unmatched uploads to fail instead.

Serialization

Criteria expose to_dict() for JSON-safe export, and load() rebuilds a criterion from a dict or JSON string. This lets rules live in config files, databases, or be exchanged across services.

import json
from koalify import F, load

rule = (F.status == "active") & (F.age >= 18) & F.role.in_("admin", "editor")

payload = rule.to_dict()
# {
#   "type": "and",
#   "criteria": [
#     {"type": "eq", "field": ["status"], "value": "active"},
#     {"type": "ge", "field": ["age"], "value": 18},
#     {"type": "in", "field": ["role"], "values": ["admin", "editor"]},
#   ],
# }

restored = load(payload)            # from dict
restored = load(json.dumps(payload)) # or from JSON string
restored(user)                       # evaluates identically

Field paths serialize as lists — attribute segments as strings, item-access segments as {"key": ...}:

(F.address.city == "London").to_dict()
# {"type": "eq", "field": ["address", "city"], "value": "London"}

(F.tags[0] == "vip").to_dict()
# {"type": "eq", "field": ["tags", {"key": 0}], "value": "vip"}

implies(...) round-trips as its Or(Not(...), ...) expansion — no special casing needed.

API

Symbol	Description
F.field	Reference a field (supports nesting: F.a.b.c and indexing: F.a[0], F.a["k"])
== != > >= < <=	Comparison operators on FieldRef
.in_(*values)	Set membership
.between(lo, hi)	Inclusive range check
&	AND (flattens nested ANDs)
|	OR (flattens nested ORs)
~	NOT
all_of(*criteria)	AND from a list
any_of(*criteria)	OR from a list
implies(a, b)	Logical implication (~a | b)
criterion.to_dict()	Serialize to a JSON-safe dict
load(data)	Rebuild a criterion from a dict or JSON string

How It Works

1. F.field_name returns a FieldRef — a lightweight path descriptor
2. Comparison operators (==, >, .in_(), etc.) produce Criterion objects
3. Criteria compose with &, |, and ~ (flattening nested groups automatically)
4. Calling a criterion resolves field values at runtime via getattr and []

Works with dataclasses, Pydantic models, namedtuples, or any object with attributes. Item access (F.tags[0], F.data["key"]) delegates to the resolved value's __getitem__, so standard IndexError / KeyError exceptions propagate naturally.

License

MIT