一日一技:在 Python 中实现延迟调用

时间:2022-11-25 16:51:34

一日一技:在 Python 中实现延迟调用

熟悉 Golang 的同学都知道,Golang 里面有一个关键词叫做defer,它可以实现延迟调用。

实际上在 Python 里面也有相关的语法,那就是contextlib.ExitStack。

我们来看这样一个场景:

我有一个函数parse,它的作用是从 Redis 中持续读入数据,并写入到MongoDB 中。示例代码如下:

  1. import json 
  2. import redis 
  3. import pymongo 
  4.  
  5. client = redis.Redis() 
  6. handler = pymongo.MongoClient().test.data 
  7.  
  8. def parse(): 
  9.     data = client.lpop('test'
  10.     if not data: 
  11.         return 
  12.     handler.insert_one(json.loads(data)) 

但现在我想增加一个需求,当Redis 读取结束或者读取数据报错的时候,能把当前的时间也写入到MongoDB 中。

那么代码可能变成下面这样:

  1. import json 
  2. import redis 
  3. import datetime 
  4. import pymongo 
  5.  
  6. client = redis.Redis() 
  7. handler = pymongo.MongoClient().test.data 
  8.  
  9. def parse(): 
  10.     while True
  11.         try: 
  12.             data = client.lpop('test'
  13.             if not data: 
  14.                 handler.insert_one({'finished'True'ts': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}) 
  15.                 return 
  16.             handler.insert_one(json.loads(data)) 
  17.         except Exception: 
  18.             handler.insert_one({'finished'True'ts': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'}) 

可以看到,代码变得很难看了。

现在,我们可以使用延迟调用来让代码变得更好看。

要实现这个目的,就可以开始使用ExitStack了。它可以注册多个回调函数,在退出上下文缩进时执行。

我们先来看一个简单的例子:

  1. import contextlib 
  2.  
  3. def callback_1(): 
  4.     print('我是第一个回调函数'
  5.  
  6. def callback_2(x): 
  7.     print(f'我是第二个回调函数,传入参数:{x}'
  8.  
  9.  
  10. with contextlib.ExitStack() as stack: 
  11.     stack.callback(callback_1) 
  12.     stack.callback(callback_2, 100) 
  13.     print(12345) 
  14.     print('xxxx'
  15. print('退出缩进'

运行效果如下图所示:

一日一技:在 Python 中实现延迟调用

可以看出以下特点:

  1. 被添加的回调函数进入了一个栈,所以后添加的回调函数先调用
  2. 回调函数会在结束缩进的时候被调用

现在我们来人工构造一个异常:

一日一技:在 Python 中实现延迟调用

可以看到,即使缩进里面出现了报错,回调函数仍然可以正常运行。等所有回调函数运行完成以后,Python 才会退出。

基于以上特点,我们就可以来重构最开始的代码了:

  1. import json 
  2. import redis 
  3. import datetime 
  4. import pymongo 
  5. import contextlib 
  6.  
  7. client = redis.Redis() 
  8. handler = pymongo.MongoClient().test.data 
  9.  
  10. def add_ts(): 
  11.     handler.insert_one({'finished'True'ts': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}) 
  12.  
  13. def parse(): 
  14.     with contextlib.ExitStack() as stack: 
  15.         stack.callback(add_ts) 
  16.         while True
  17.             data = client.lpop('test'
  18.             if not data: 
  19.                 return 
  20.             handler.insert_one(json.loads(data)) 

无论是正常运行结束还是运行过程中报错,add_ts函数都会正常运行,确保始终增加一条日期数据。

原文链接:https://mp.weixin.qq.com/s/fjbOA0YFW1jrsthtNem5aQ