与 fastapi 并行执行一个函数

Question

我想执行一个与快速 api 并行运行的函数。

该功能基本上是从网络上获取数据并将其写入数据库，数据可能是百万条记录，可能需要超过 5 分钟。

API 接受数据并查询数据库并响应客户端。

我现在正在做的是

@app.on_event("startup")
async def startup_event():
    
    await Extractor().callit()
class Extractor:
    async def callit(self):
        # Insertion of the tor ips with the data models
        indicators = []
        indicators.clear()
        for ip in await scrap_web_ips():
            if ip == '':
                continue
            q_indicator = self.check_ip_status(ip)
            if q_indicator is None:
                indicator = Indicators(indicator=ip, tor=True,
                                       last_seen=get_current_date_in_iso(),
                                       first_seen=get_current_date_in_iso(),
                                       white_list=True if self.get_vt_stats(ip) == 0 else False)
                indicators.append(indicator)
            else:
                print("Updating the lastseen of the indicator.")
                q_indicator.last_seen = get_current_date_in_iso()
                q_indicator.tor = True
                self.session.commit()
    

         print(self.insert_data(indicators))

还有我的 API。

该函数正在阻止 api 调用。

我现在不知道如何处理。

Answer 1

callit() 函数很可能实际上根本不是异步函数。你说它从网上收集一些东西并将其存储在数据库中。这两个动作是使用异步方法的主要候选者。这看起来像这样：

async def callit():
1  result = await get_from_web_async()
2  await store_in_db_async(result)

在第 1 行，它会在等待 get_from_web_async() 的结果时返回控制权，而在第 2 行，它会在将其存储在数据库中时再次返回控制权。

但是，如果没有你在 callit() 函数上的代码，这只是一个猜测 :)

编辑： 根据附加信息，工作原理是对scrap_web_ips() 的调用不返回异步生成器。因此，在循环之前等待整个结果。虽然在等待结果时调用没有阻塞，但它所使用的循环将完全阻塞。