Reputation: 31
Here is the Python code:
def _get_handler_by_topic_arn(topic_arn: str, event_name: str, event_message: dict):
if topic_arn == CONFIG.get("MT_MAIN_SNS_TOPIC_ARN"):
return MT_MAIN_TOPIC_HANDLERS.get(event_name)
if topic_arn == CONFIG.get("FOX_REQUEST_SNS_TOPIC_ARN"):
return FOX_REQUEST_TOPIC_HANDLERS.get(event_name)
if topic_arn == CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN"):
if event_message.get("status") and event_message.get("status") != "CANCELLED":
return None
return FOX_ORDER_RESULTS_TOPIC_HANDLERS.get(event_name)
return None
It works but I'm searching for a more efficient / elegant / simplest way to do the same. If you have any idea, I'll take it :).
Upvotes: 1
Views: 54
Reputation: 1537
Here's an approach with a handler
map and a validator
map to decouple the validation logic from the handling logic:
HANDLER_MAP = {
CONFIG.get("MT_MAIN_SNS_TOPIC_ARN"): MT_MAIN_TOPIC_HANDLERS,
CONFIG.get("FOX_REQUEST_SNS_TOPIC_ARN"): FOX_REQUEST_TOPIC_HANDLERS,
CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN"): FOX_ORDER_RESULTS_TOPIC_HANDLERS
}
VALIDATOR_MAP = {
CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN"): lambda message: message.get("status") != "CANCELLED"
}
def get_handler_and_validator(topic_arn: str) -> tuple[dict, Callable]:
handler = HANDLER_MAP.get(topic_arn, {})
validator = VALIDATOR_MAP.get(topic_arn, lambda _: True)
return handler, validator
def _get_handler_by_topic_arn(topic_arn: str, event_name: str, event_message: dict):
topic_handler, validator = get_handler_and_validator(topic_arn)
if validator(event_message):
return topic_handler.get(event_name)
Upvotes: 0
Reputation: 1642
You can use a dictionary to avoid the if
s:
def _get_handler_by_topic_arn(topic_arn: str, event_name: str, event_message: dict):
not_canceled = event_message.get("status") and event_message.get("status") != "CANCELLED"
handlers = {
CONFIG.get("MT_MAIN_SNS_TOPIC_ARN"): MT_MAIN_TOPIC_HANDLERS.get(event_name),
CONFIG.get("FOX_REQUEST_SNS_TOPIC_ARN"): FOX_REQUEST_TOPIC_HANDLERS.get(event_name),
CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN"): None if not_canceled else FOX_ORDER_RESULTS_TOPIC_HANDLERS.get(event_name)
}
return handlers.get(topic_arn)
Since the function can also return None
if the status is not CANCELLED
, the dictionary takes that into account when setting the value for the key CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN")
.
Also, you can customize what will be returned if the topic_arn
doesn't exist. To do that, just pass the default return to handlers.get(topic_arn)
, like so:
handlers.get(topic_arn, "NOT FOUND")
Upvotes: 2
Reputation: 152
you could also do something like
topics = {
"MT_MAIN_SNS_TOPIC_ARN" : MT_MAIN_TOPIC_HANDLERS,
"FOX_REQUEST_SNS_TOPIC_ARN" : FOX_REQUEST_TOPIC_HANDLERS,
"FOX_ORDER_RESULTS_SNS_TOPIC_ARN" : FOX_ORDER_RESULTS_TOPIC_HANDLERS
}
def _get_handler_by_topic_arn(topic_arn: str, event_name: str, event_message: dict):
for k,v in topics.items():
if topic_arn == CONFIG.get(k):
return v.get(event_name)
Upvotes: 1
Reputation: 2082
I would refactor your code like this, so it's less crowded and also there is only one return point (so the control flow is easier to examine).
def _get_handler_by_topic_arn(topic_arn: str, event_name: str, event_message: dict):
ret_value = None
# Give a better name to conds
cond1 = topic_arn == CONFIG.get("MT_MAIN_SNS_TOPIC_ARN")
cond2 = topic_arn == CONFIG.get("FOX_REQUEST_SNS_TOPIC_ARN")
cond3 = topic_arn == CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN")
cond4 = event_message.get("status") and event_message.get("status") != "CANCELLED"
if cond1:
ret_value = MT_MAIN_TOPIC_HANDLERS.get(event_name)
elif cond2:
ret_value = FOX_REQUEST_TOPIC_HANDLERS.get(event_name)
elif cond3:
if cond4:
ret_value = None
else:
ret_value = FOX_ORDER_RESULTS_TOPIC_HANDLERS.get(event_name)
return ret_value
Upvotes: 1