Filters

There are a few available filters to help you make sure the correct processor is invoked for the correct event. To see how to use filters in practice, see the core concepts.

Warning

It’s possible to create different filters that will match the same event. For example, when using the Exists and Eq filters on the same key, if the Eq filter matches, then the Exists filter is guaranteed to match.

Have a look at Ranking Processors to learn how to resolve these ambiguities. Also, note that these issues may not apply to your context. You only have to worry about this if you have ambiguous filters.

Accept

This filter will always match any event it is presented with. It will even match things that are not dictionaries. Use this if you need to take a default action whenever no processor exists for an event, or if an unexpected event was sent to your system.

from event_processor.filters import Accept

accept = Accept()

print(accept.matches({}))
print(accept.matches(None))
print(accept.matches({"Hello", "World"}))

True
True
True

Exists

This filter matches events that contain a certain key (which can be nested), but the value can be anything.

from event_processor.filters import Exists

a_exists = Exists("a")
nested = Exists("a.b.c")

print(a_exists.matches({"a": None}))
print(a_exists.matches({"a": 2}))
print(a_exists.matches({}))
print(nested.matches({"a": {"b": {"c": None}}}))
print(nested.matches({"a": {"b": {"c": 0}}}))

True
True
False
True
True

Eq

This filter matches a subset of the events matched by Exists. It only matches the events where a specific value is found at the specified key (as opposed to just existing).

from event_processor.filters import Eq

a_is_b = Eq("a", "b")
a_b_c_is_none = Eq("a.b.c", None)

print(a_is_b.matches({"a": "b"}))
print(a_is_b.matches({"a": 2}))
print(a_b_c_is_none.matches({"a": {"b": {"c": None}}}))
print(a_b_c_is_none.matches({"a": {"b": {"c": 0}}}))

True
False
True
False

And

This filter does exactly what you would expect, and matches when all the events supplied to it as arguments match. It acts as a logical AND between all its sub-filters.

from event_processor.filters import And, Exists

a_exists = Exists("a")
b_exists = Exists("b")
c_exists = Exists("c")

a_and_b_exist = And(a_exists, b_exists)
a_b_and_c_exist = And(a_exists, b_exists, c_exists)

print(a_and_b_exist.matches({"a": 0, "b": 0}))
print(a_and_b_exist.matches({"a": 0, "b": 0, "c": 0}))
print(a_b_and_c_exist.matches({"a": 0, "b": 0}))
print(a_b_and_c_exist.matches({"a": 0, "b": 0, "c": 0}))

True
True
False
True

You can also use & between processors instead of And explicitly to make your filters prettier.

from event_processor.filters import And, Exists

a_exists = Exists("a")
b_exists = Exists("b")
c_exists = Exists("c")

a_and_b_exist = a_exists & b_exists
a_b_and_c_exist = a_exists & b_exists & c_exists

print(a_and_b_exist.matches({"a": 0, "b": 0}))
print(a_and_b_exist.matches({"a": 0, "b": 0, "c": 0}))
print(a_b_and_c_exist.matches({"a": 0, "b": 0}))
print(a_b_and_c_exist.matches({"a": 0, "b": 0, "c": 0}))

True
True
False
True

Or

This filter is similar to the And filter, except that it will match if any of its sub-filters match.

from event_processor.filters import Or, Exists

a_exists = Exists("a")
b_exists = Exists("b")
c_exists = Exists("c")

a_b_or_c_exist = Or(a_exists, b_exists, c_exists)

print(a_b_or_c_exist.matches({"a": 0}))
print(a_b_or_c_exist.matches({"b": 0}))
print(a_b_or_c_exist.matches({"c": 0}))
print(a_b_or_c_exist.matches({"d": 0}))

True
True
True
False

Again, to make things more ergonomic, you can use | instead of Or.

from event_processor.filters import Or, Exists

a_exists = Exists("a")
b_exists = Exists("b")
c_exists = Exists("c")

a_b_or_c_exist = a_exists | b_exists | c_exists

print(a_b_or_c_exist.matches({"a": 0}))
print(a_b_or_c_exist.matches({"b": 0}))
print(a_b_or_c_exist.matches({"c": 0}))
print(a_b_or_c_exist.matches({"d": 0}))

True
True
True
False