Python中使用OSMnx如何捕获Overpass API日志并实现重试？

1. 自定义日志处理器捕获目标日志 #

OSMnx内部基于Python标准库的logging模块输出日志，我们可以创建自定义日志Handler，精准捕获Overpass的内存不足类运行时错误：

import logging
import osmnx as ox

# 自定义异常类，用于标记Overpass内存不足错误
class OverpassMemoryError(Exception):
    pass

# 自定义日志处理器
class OverpassLogHandler(logging.Handler):
    def emit(self, record):
        log_content = self.format(record)
        # 匹配目标错误日志的特征字符串
        if "runtime error: Query run out of memory using about" in log_content:
            # 抛出异常，供重试逻辑识别
            raise OverpassMemoryError(log_content)

# 获取OSMnx的日志实例
ox_logger = logging.getLogger(osmnx.__name__)
# 添加自定义处理器
custom_handler = OverpassLogHandler()
ox_logger.addHandler(custom_handler)
# 设置日志级别，确保能捕获到目标日志（根据实际输出级别调整，比如ERROR/WARNING）
ox_logger.setLevel(logging.ERROR)

2. 结合重试装饰器实现自动重试 #

借助tenacity库（需提前安装：pip install tenacity），给你的OSMnx查询函数添加重试逻辑，捕获自定义异常时触发重试：

from tenacity import retry, stop_after_attempt, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    retry=retry_if_exception_type(OverpassMemoryError)
)
def execute_osmnx_query():
    # 替换为你的实际OSMnx查询代码
    graph = ox.graph_from_place("Shanghai, China", network_type="walk")
    return graph

try:
    final_result = execute_osmnx_query()
except OverpassMemoryError as err:
    print(f"多次重试后仍失败: {err}")

3. 关键注意事项 #

• 日志级别要匹配：如果目标日志是WARNING级别，需将ox_logger.setLevel改为logging.WARNING，确保能捕获到对应日志。
• 用完处理器记得移除：如果后续不需要该逻辑，执行ox_logger.removeHandler(custom_handler)避免干扰其他日志输出。
• 可扩展重试策略：比如添加指数退避等待（wait=wait_exponential(multiplier=1, min=2, max=10)），避免频繁请求Overpass API。

内容的提问来源于stack exchange，提问作者beatrice