求更简洁优雅的Python条件返回处理器实现方案
更高效优雅的Topic ARN事件处理器匹配实现
优化方案1:用映射字典统一管理逻辑
def _get_handler_by_topic_arn(topic_arn: str, event_name: str, event_message: dict): # 建立topic ARN与处理器、过滤条件的映射 topic_map = { CONFIG.get("MT_MAIN_SNS_TOPIC_ARN"): (MT_MAIN_TOPIC_HANDLERS, lambda msg: True), CONFIG.get("FOX_REQUEST_SNS_TOPIC_ARN"): (FOX_REQUEST_TOPIC_HANDLERS, lambda msg: True), CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN"): ( FOX_ORDER_RESULTS_TOPIC_HANDLERS, lambda msg: not (msg.get("status") and msg.get("status") != "CANCELLED") ) } handler_info = topic_map.get(topic_arn) if handler_info: handlers, check_condition = handler_info if check_condition(event_message): return handlers.get(event_name) return None
优化方案2:全局初始化映射(性能更优)
如果该函数会被频繁调用,建议把映射移到函数外初始化,避免重复创建字典:
# 全局初始化topic配置映射(仅执行一次) TOPIC_HANDLER_MAPPING = { CONFIG.get("MT_MAIN_SNS_TOPIC_ARN"): { "handlers": MT_MAIN_TOPIC_HANDLERS, "condition": lambda msg: True }, CONFIG.get("FOX_REQUEST_SNS_TOPIC_ARN"): { "handlers": FOX_REQUEST_TOPIC_HANDLERS, "condition": lambda msg: True }, CONFIG.get("FOX_ORDER_RESULTS_SNS_TOPIC_ARN"): { "handlers": FOX_ORDER_RESULTS_TOPIC_HANDLERS, "condition": lambda msg: not (msg.get("status") and msg.get("status") != "CANCELLED") } } def _get_handler_by_topic_arn(topic_arn: str, event_name: str, event_message: dict): topic_config = TOPIC_HANDLER_MAPPING.get(topic_arn) if topic_config and topic_config["condition"](event_message): return topic_config["handlers"].get(event_name) return None
优化思路说明
- 替换链式if判断:用字典映射实现O(1)时间复杂度的查找,比原代码的O(n)更高效,后续新增topic时只需在映射中添加条目,无需修改函数逻辑
- 统一处理过滤条件:把每个topic的特殊判断逻辑封装成条件函数,与处理器绑定,逻辑更清晰,扩展性更强
- 减少重复操作:方案2将映射全局初始化,避免每次调用函数都重新构建字典,进一步提升性能
内容的提问来源于stack exchange,提问作者Asthonishia
相关产品推荐
相关产品推荐

