如何复用同步函数与异步协程的节流检查逻辑？

我编写了如下同步函数：

def check_for_throttling(r):
    if r.status_code == responses.THROTTLING:
        # <lots of other stuff>
        time.sleep(Throttle.DELAY)
        Throttle.DELAY = Throttle.DELAY + 2
    return r

同时也编写了如下异步协程：

async def check_for_throttling(r):
    if r.status_code == responses.THROTTLING:
        # <lots of other stuff>
        await asyncio.sleep(Throttle.DELAY)
        Throttle.DELAY = Throttle.DELAY + 2
    return r

二者逻辑几乎完全一致，仅sleep方式不同（一个用time.sleep()，另一个用asyncio.sleep()）。我还有其他逻辑更复杂的类似场景，请问是否有办法复用同一代码块处理这两种场景？

可以通过拆分核心逻辑与IO操作的方式来复用代码，以下是几种实用方案：

方案1：提取纯逻辑代码块（推荐） #

把两个函数中除了sleep之外的核心业务逻辑单独抽成通用函数，同步和异步版本仅处理各自的sleep调用：

首先提取纯业务逻辑：

def _handle_throttling():
    # 这里放原代码里的<lots of other stuff>
    current_delay = Throttle.DELAY
    Throttle.DELAY += 2
    return current_delay

同步版本直接调用：

import time

def check_for_throttling(r):
    if r.status_code == responses.THROTTLING:
        delay = _handle_throttling()
        time.sleep(delay)
    return r

异步版本则用await调用异步sleep：

import asyncio

async def check_for_throttling(r):
    if r.status_code == responses.THROTTLING:
        delay = _handle_throttling()
        await asyncio.sleep(delay)
    return r

这种方式让90%的核心逻辑完全复用，只剩一行sleep的差异，简洁直观，无需额外依赖。

方案2：依赖注入sleep函数 #

如果想让逻辑结构更统一，可以把sleep操作作为参数传入通用逻辑函数，同步传time.sleep，异步则封装成可await的调用：

先编写通用协程逻辑：

async def _check_core(r, sleep_func):
    if r.status_code == responses.THROTTLING:
        # <lots of other stuff>
        await sleep_func(Throttle.DELAY)
        Throttle.DELAY += 2
    return r

同步版本用asyncio.run包装调用，同时把同步sleep适配成可await的函数：

import time
import asyncio

def check_for_throttling(r):
    async def sync_sleep_wrapper(delay):
        time.sleep(delay)
    return asyncio.run(_check_core(r, sync_sleep_wrapper))

异步版本直接使用：

import asyncio

async def check_for_throttling(r):
    return await _check_core(r, asyncio.sleep)

这种方式适合逻辑特别复杂、不想重复写判断分支的场景，但同步版本需要额外适配sleep函数，稍显繁琐。

方案3：使用统一IO库（可选） #

如果项目允许引入第三方库，anyio提供了统一的同步/异步IO接口，它的anyio.sleep在两种环境下都能正常工作，只需写一份核心代码：

import anyio

async def check_for_throttling(r):
    if r.status_code == responses.THROTTLING:
        # <lots of other stuff>
        await anyio.sleep(Throttle.DELAY)
        Throttle.DELAY += 2
    return r

# 同步调用时用anyio.run包装
def sync_check_for_throttling(r):
    return anyio.run(check_for_throttling, r)

这种方式代码最简洁，但需要引入外部依赖，适合已经使用异步生态的项目。

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