几年前，在 StackOverflow 上看到一个问题让我印象深刻：C++ 有 __FILE__ 和 __LINE__，但 Python 只有 __file__，没有类似的 __line__。是否可以实现类似的功能？当时的答案通常是通过 inspect 获取行号，但这需要通过函数调用。于是，我想到了通过实现 __str__ 方法来实现与 C++ 相同的效果：

import inspect

class LineNo:
    """动态获取当前调用行号的类。"""
    def __str__(self):
        # f_back 返回上一帧，f_lineno 返回行号
        return str(inspect.currentframe().f_back.f_lineno)

__line__ = LineNo()

print(f"当前代码位于第 {__line__} 行。") 
print(f"另一个检查点在第 {__line__} 行。")

LineNo 的实例可以像字符串一样使用，__line__ 的使用方式也避免了函数调用，看起来就像是内建变量。

这几年由于工作原因我需要编写很多下载文件的函数，这些操作非常耗时，但下载结果不会被立即使用。虽然 Python 提供了丰富的多线程的 API，但如果下载调用分散在各处，即使用多线程这些语句仍会阻塞 CPU，浪费了处理数据的时间。

我想到 Python 的 future 机制，如果只是简单返回 future，使用者需要调用 result()，这样会改变函数的用法。这时我想到了 __str__ 的灵感：我们是否可以返回一个对象实例，这个对象可以被用来当作文件路径？

Python 提供了一个路径协议（__fspath__），只要对象实现了这个方法，它就能被 open() 这样的函数识别为文件路径。我们利用这一点，将等待下载完成的阻塞操作延迟到 __fspath__ 被调用时执行。 机制是这样的：

  1. 调用下载函数时，下载任务立即被扔到线程池启动。
  2. 函数立即返回一个特殊的 AsyncDownloadPath 对象。
  3. 程序可以继续执行其他任务。
  4. 只有当程序试图用类似 open 的方法访问这个对象（文件路径）时，系统才会调用对象的 __fspath__ 方法。
  5. 在 __fspath__ 中，程序才会被阻塞，等待后台的下载任务真正完成。 我们将它封装成一个装饰器：
import concurrent.futures
from concurrent.futures import Future
import functools
import atexit
import time
from typing import Callable, TypeVar, ParamSpec

R = TypeVar("R")
P = ParamSpec("P")

class AsyncDownloadPath:
    _executor = concurrent.futures.ThreadPoolExecutor()
    
    def __init__(self, download_func: Callable[[], str]):
        self._future: Future[str] = self._executor.submit(download_func)
        self._path: str | None = None

    def __fspath__(self) -> str:
        """当需要文件路径时触发下载，阻塞直到完成。"""
        if not self._path:
            print("⏳ 触发 __fspath__，等待下载完成...")
            self._path = self._future.result()
            print("✅ 下载完成。")
        return self._path

    @classmethod
    def shutdown(cls):
        cls._executor.shutdown(wait=False)

def async_download(func: Callable[P, str]) -> Callable[P, AsyncDownloadPath]:
    """将普通函数转换为异步下载路径的装饰器。"""
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> AsyncDownloadPath:
        return AsyncDownloadPath(lambda: func(*args, **kwargs))
    return wrapper

atexit.register(AsyncDownloadPath.shutdown)

通过上面的实现，我们可以轻松地将一个阻塞的文件下载函数转变为异步下载：

@async_download
def prepare_heavy_file(delay):
    time.sleep(delay) 
    return "/tmp/downloaded_data.log"

my_file = prepare_heavy_file(1) # 函数立即返回
print("任务已提交，继续其他操作...")

# 在此触发 __fspath__，下载完成后再执行
with open(my_file, 'w') as f:
    f.write("Processing.")

注意： 下载文件的函数不能抛出异常，可以通过返回 None 来表示下载失败。

由于 Python 的灵活性，这种方法可以用在几乎任意耗时的任务中。如果一个对象的创建很耗时，可以把创建这个对象的过程放在线程中执行并立刻返回一个Proxy对象。Python 可以通过__getattr__和 __setattr__等方法很方便的实现Proxy模式，并只在访问对象属性的时候进行阻塞。甚至连import语句也是可以放在后台线程执行的（通过重写builtins.__import__方法）,import torch消耗的时间完全可以做一些类似读数据的操作。

魔术方法的应用远不止这些。我曾见过将 __ror__ 和 __gt__ 结合使用，创造出简洁的 data | command | command > file 语法。Python 中或许还有许多值得挖掘的用法等待我们发现。