SNS:subscribe() now has increased support for the FilterPolicy-argument (#5436)
This commit is contained in:
Bert Blommers
GitHub
parent
89008c4230
commit
d145471b3f
@ -265,8 +265,6 @@ class Subscription(BaseModel):
|
||||
)
|
||||
|
||||
def _matches_filter_policy(self, message_attributes):
|
||||
# TODO: support Anything-but matching, prefix matching and
|
||||
# numeric value matching.
|
||||
if not self._filter_policy:
|
||||
return True
|
||||
|
||||
@ -311,12 +309,75 @@ class Subscription(BaseModel):
|
||||
return True
|
||||
if isinstance(rule, dict):
|
||||
keyword = list(rule.keys())[0]
|
||||
attributes = list(rule.values())[0]
|
||||
value = list(rule.values())[0]
|
||||
if keyword == "exists":
|
||||
if attributes and field in message_attributes:
|
||||
if value and field in message_attributes:
|
||||
return True
|
||||
elif not attributes and field not in message_attributes:
|
||||
elif not value and field not in message_attributes:
|
||||
return True
|
||||
elif keyword == "prefix" and isinstance(value, str):
|
||||
if field in message_attributes:
|
||||
attr = message_attributes[field]
|
||||
if attr["Type"] == "String" and attr["Value"].startswith(
|
||||
value
|
||||
):
|
||||
return True
|
||||
elif keyword == "anything-but":
|
||||
if field not in message_attributes:
|
||||
continue
|
||||
attr = message_attributes[field]
|
||||
if isinstance(value, dict):
|
||||
# We can combine anything-but with the prefix-filter
|
||||
anything_but_key = list(value.keys())[0]
|
||||
anything_but_val = list(value.values())[0]
|
||||
if anything_but_key != "prefix":
|
||||
return False
|
||||
if attr["Type"] == "String":
|
||||
actual_values = [attr["Value"]]
|
||||
else:
|
||||
actual_values = [v for v in attr["Value"]]
|
||||
if all(
|
||||
[
|
||||
not v.startswith(anything_but_val)
|
||||
for v in actual_values
|
||||
]
|
||||
):
|
||||
return True
|
||||
else:
|
||||
undesired_values = (
|
||||
[value] if isinstance(value, str) else value
|
||||
)
|
||||
if attr["Type"] == "Number":
|
||||
actual_values = [str(attr["Value"])]
|
||||
elif attr["Type"] == "String":
|
||||
actual_values = [attr["Value"]]
|
||||
else:
|
||||
actual_values = [v for v in attr["Value"]]
|
||||
if all([v not in undesired_values for v in actual_values]):
|
||||
return True
|
||||
elif keyword == "numeric" and isinstance(value, list):
|
||||
# [(< x), (=, y), (>=, z)]
|
||||
numeric_ranges = zip(value[0::2], value[1::2])
|
||||
if (
|
||||
message_attributes.get(field, {}).get("Type", "")
|
||||
== "Number"
|
||||
):
|
||||
msg_value = message_attributes[field]["Value"]
|
||||
matches = []
|
||||
for operator, test_value in numeric_ranges:
|
||||
test_value = int(test_value)
|
||||
if operator == ">":
|
||||
matches.append((msg_value > test_value))
|
||||
if operator == ">=":
|
||||
matches.append((msg_value >= test_value))
|
||||
if operator == "=":
|
||||
matches.append((msg_value == test_value))
|
||||
if operator == "<":
|
||||
matches.append((msg_value < test_value))
|
||||
if operator == "<=":
|
||||
matches.append((msg_value <= test_value))
|
||||
return all(matches)
|
||||
attr = message_attributes[field]
|
||||
return False
|
||||
|
||||
return all(
|
||||
|
||||
@ -1120,3 +1120,165 @@ def test_filtering_all_AND_matching_no_match():
|
||||
message_bodies.should.equal([])
|
||||
message_attributes = [json.loads(m.body)["MessageAttributes"] for m in messages]
|
||||
message_attributes.should.equal([])
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_prefix():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"prefix": "bas"}]}
|
||||
)
|
||||
|
||||
for interest, idx in [("basketball", "1"), ("rugby", "2"), ("baseball", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "String", "StringValue": interest},
|
||||
},
|
||||
)
|
||||
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
set(message_bodies).should.equal({"match1", "match3"})
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_anything_but():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"anything-but": "basketball"}]}
|
||||
)
|
||||
|
||||
for interest, idx in [("basketball", "1"), ("rugby", "2"), ("baseball", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "String", "StringValue": interest},
|
||||
},
|
||||
)
|
||||
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
set(message_bodies).should.equal({"match2", "match3"})
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_anything_but_multiple_values():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"anything-but": ["basketball", "rugby"]}]}
|
||||
)
|
||||
|
||||
for interest, idx in [("basketball", "1"), ("rugby", "2"), ("baseball", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "String", "StringValue": interest},
|
||||
},
|
||||
)
|
||||
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
set(message_bodies).should.equal({"match3"})
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_anything_but_prefix():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"anything-but": {"prefix": "bas"}}]}
|
||||
)
|
||||
|
||||
for interest, idx in [("basketball", "1"), ("rugby", "2"), ("baseball", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "String", "StringValue": interest},
|
||||
},
|
||||
)
|
||||
|
||||
# This should match rugby only
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
set(message_bodies).should.equal({"match2"})
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_anything_but_unknown():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"anything-but": {"unknown": "bas"}}]}
|
||||
)
|
||||
|
||||
for interest, idx in [("basketball", "1"), ("rugby", "2"), ("baseball", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "String", "StringValue": interest},
|
||||
},
|
||||
)
|
||||
|
||||
# This should match rugby only
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
message_bodies.should.equal([])
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_anything_but_numeric():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"anything-but": ["100"]}]}
|
||||
)
|
||||
|
||||
for nr, idx in [("50", "1"), ("100", "2"), ("150", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "Number", "StringValue": nr},
|
||||
},
|
||||
)
|
||||
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
set(message_bodies).should.equal({"match1", "match3"})
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_numeric_match():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"numeric": ["=", "100"]}]}
|
||||
)
|
||||
|
||||
for nr, idx in [("50", "1"), ("100", "2"), ("150", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "Number", "StringValue": nr},
|
||||
},
|
||||
)
|
||||
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
set(message_bodies).should.equal({"match2"})
|
||||
|
||||
|
||||
@mock_sqs
|
||||
@mock_sns
|
||||
def test_filtering_numeric_range():
|
||||
topic, queue = _setup_filter_policy_test(
|
||||
{"customer_interests": [{"numeric": [">", "49", "<=", "100"]}]}
|
||||
)
|
||||
|
||||
for nr, idx in [("50", "1"), ("100", "2"), ("150", "3")]:
|
||||
topic.publish(
|
||||
Message=f"match{idx}",
|
||||
MessageAttributes={
|
||||
"customer_interests": {"DataType": "Number", "StringValue": nr},
|
||||
},
|
||||
)
|
||||
|
||||
messages = queue.receive_messages(MaxNumberOfMessages=5)
|
||||
message_bodies = [json.loads(m.body)["Message"] for m in messages]
|
||||
set(message_bodies).should.equal({"match1", "match2"})
|
||||
|
||||
Loading…
Reference in New Issue
Block a user