Python requests与JSON数据刷新问题求助:API定时数据无法实时获取
看起来你遇到的核心问题是变量作用域导致的——你的refresh_data()函数里的data是局部变量,根本没更新全局的data,所以循环里打印的一直是初始值。另外你提到需要在多个地方调用刷新逻辑,我给你几个从简单到进阶的优化方案,你可以根据自己的程序复杂度选择:
方案1:修正全局变量作用域(快速解决)
如果你的程序比较简单,用全局变量能快速搞定,但要注意全局变量在复杂项目里容易引发耦合问题:
import requests import time url = 'https://api.url.com/...' data = {} # 先初始化全局变量 def refresh_data(): global data # 声明要修改的是全局变量 try: resp = requests.get(url) resp.raise_for_status() # 主动抛出HTTP错误 data = resp.json() except Exception as e: print(f"刷新数据失败: {str(e)}") # 可选:保留旧数据或者设置默认值 pass # 程序启动先刷新一次 refresh_data() while True: print(data) refresh_data() time.sleep(300) # 加个间隔,避免频繁请求API被限流
方案2:让函数返回新数据(更灵活)
把刷新逻辑和数据存储解耦,函数只负责获取最新数据,调用方自己处理数据存储。这种方式更适合在多个地方调用刷新逻辑的场景:
import requests import time url = 'https://api.url.com/...' def refresh_data(): try: resp = requests.get(url) resp.raise_for_status() return resp.json() except Exception as e: print(f"刷新数据失败: {str(e)}") return None # 或者返回上次的旧数据,需要调用方处理 # 初始化数据 data = refresh_data() while True: print(data) new_data = refresh_data() if new_data is not None: data = new_data time.sleep(300) # 其他需要刷新的地方直接调用即可 def process_data(): latest_data = refresh_data() if latest_data: # 处理数据逻辑 pass
方案3:用类封装(适合复杂项目)
如果你的程序后续要扩展功能(比如缓存、重试、多数据源),用类封装数据和刷新逻辑会更易维护:
import requests import time from datetime import datetime, timedelta class APIDataFetcher: def __init__(self, url, cache_duration=300): self.url = url self.cache_duration = cache_duration # 缓存有效期(秒) self._data = None self._last_refresh_time = None def _fetch_from_api(self): try: resp = requests.get(self.url) resp.raise_for_status() return resp.json() except Exception as e: print(f"API请求失败: {str(e)}") return None def get_latest_data(self): # 检查缓存是否过期,没过期就返回缓存数据 current_time = datetime.now() if (self._last_refresh_time is None or current_time - self._last_refresh_time > timedelta(seconds=self.cache_duration)): new_data = self._fetch_from_api() if new_data is not None: self._data = new_data self._last_refresh_time = current_time return self._data # 使用示例 fetcher = APIDataFetcher('https://api.url.com/...') while True: print(fetcher.get_latest_data()) time.sleep(60) # 其他地方调用同样的实例即可获取最新/缓存数据 def analyze_data(): data = fetcher.get_latest_data() if data: # 分析逻辑 pass
额外的最佳实践
- 一定要加错误处理:API请求可能因为网络、权限、格式错误失败,用
try-except捕获异常避免程序崩溃 - 控制请求频率:不要无间隔刷新,按照API文档的限制设置合理的间隔时间
- 可选添加重试机制:请求失败时可以重试1-2次,提升稳定性
内容的提问来源于stack exchange,提问作者Stivan
相关产品推荐
相关产品推荐

