Python使用propcache缓存属性与异步接口_Python

1. 安装与导入：一行命令搞定 propcache

propcache 是一个专注于属性级缓存的轻量库，核心价值在于为类属性提供“惰性求值 + 自动失效”的能力。它常与 aiohttp 搭配使用——实际上，aiohttp 内部就依赖 propcache 来缓存请求头、响应体等解析结果。先安装它：

pip install propcache

如果你在写异步 web 服务，大概率已经间接装过它了。验证导入并查看版本：

import propcache

print(propcache.__version__)  # 例如 1.2.0
print(hasattr(propcache, "cached_property"))  # true

propcache 的核心 api 只有两个：cached_property 和 under_cached_property。前者是装饰器，用于将类方法变成“只计算一次”的属性；后者则用于在类内部定义时标记“这个属性由缓存管理”。我们先看最常用的 cached_property：

from propcache import cached_property

class dataloader:
    def __init__(self, raw_data: str):
        self._raw = raw_data

    @cached_property
    def parsed(self) -> list[dict]:
        # 模拟昂贵的解析操作
        print("解析中……只执行一次")
        return [{"row": line} for line in self._raw.split(",")]

loader = dataloader("a,b,c")
print(loader.parsed)   # 触发计算
print(loader.parsed)   # 直接返回缓存，不再打印

关键语法点：

@cached_property 必须装饰在实例方法上，且方法只接收 self，不能有额外参数。
首次访问时执行方法体，返回值被缓存到实例的 __dict__ 中；后续访问直接读取缓存，不再调用原方法。
缓存是每实例独立的，不同实例互不干扰。
与标准库 functools.cached_property 的区别在于：propcache 的版本针对 aiohttp 内部做了优化，并且支持异步属性——这是它的最大卖点。

再看异步场景。cached_property 同样支持 async def 方法，但访问方式不同：

from propcache import cached_property
import asyncio

class asyncconfig:
    def __init__(self, url: str):
        self.url = url

    @cached_property
    async def data(self) -> dict:
        # 模拟异步 http 请求
        await asyncio.sleep(0.1)
        return {"status": "ok", "url": self.url}

async def main():
    cfg = asyncconfig("https://example.com")
    # 注意：异步属性必须用 await 获取
    first = await cfg.data
    second = await cfg.data
    print(first is second)  # true，缓存生效

asyncio.run(main())

这里有个易错点：异步 cached_property 的返回值是协程对象，必须用 await 解包。首次 await 时执行方法体并缓存结果；第二次 await 时直接返回缓存值，不会重复执行网络请求。如果你忘记 await，会拿到一个 coroutine 对象，导致逻辑错误。

propcache 还提供一个 under_cached_property，它不直接参与计算，而是声明“某个属性由外部缓存控制”，通常用于框架内部标记。日常业务代码中，你几乎只需要 cached_property。

最后提醒：不要对可变对象使用 cached_property 后直接原地修改，否则缓存中的值会变化，但不会触发重新计算。如果属性依赖外部状态且可能失效，请手动删除实例 __dict__ 中的对应键：

loader = dataloader("x,y")
_ = loader.parsed          # 触发计算
del loader.__dict__["parsed"]  # 手动失效
print(loader.parsed)       # 重新计算

安装至此结束。下一节我们深入 cached_property 的源码级行为，理解它如何避免线程竞争，以及为什么在 aiohttp 中能显著降低请求延迟。

2. 核心对象与语法：cached_property 与 async_cached_property

安装 propcache 只需一行命令，它不依赖任何第三方库，适合直接嵌入现有项目：

pip install propcache

导入时，最常用的是两个装饰器：cached_property 用于同步类，async_cached_property 用于异步类（即类的方法定义为 async def）。它们的用法几乎一致，区别仅在于被装饰的函数是否为协程。

同步场景：cached_property

cached_property 是一个非数据描述符，作用是把一个实例方法变成“只计算一次”的属性。首次访问时执行方法体，之后每次访问都直接返回缓存结果，不再重复调用原函数。

from propcache import cached_property

class dataloader:
    def __init__(self, source: str):
        self.source = source

    @cached_property
    def raw_data(self) -> list[int]:
        # 模拟耗时 io，比如读取大文件或请求远端 api
        print(f"正在从 {self.source} 加载数据...")
        return list(range(10_000))

loader = dataloader("demo.txt")
print("第一次访问:")
print(len(loader.raw_data))
print("第二次访问:")
print(len(loader.raw_data))

执行结果中，正在从 demo.txt 加载数据... 只会打印一次。第二次访问 loader.raw_data 时，propcache 直接从实例的 __dict__ 中取值，不再进入方法体。注意：装饰器要求被装饰函数不能有参数（除了 self），返回值会被原样缓存——如果返回值是可变对象（如列表、字典），外部修改会直接影响缓存内容，后续访问拿到的是修改后的对象。

异步场景：async_cached_property

当属性计算涉及 await（比如异步 http 请求、数据库查询）时，必须使用 async_cached_property。被装饰的方法需要定义为 async def，访问方式也有所不同——你不能直接 await instance.attr，而是需要先获取协程对象再等待。

import asyncio
from propcache import async_cached_property

class asyncuserinfo:
    def __init__(self, user_id: int):
        self.user_id = user_id

    @async_cached_property
    async def profile(self) -> dict:
        # 模拟异步数据库查询
        await asyncio.sleep(0.1)
        return {"id": self.user_id, "name": "alice"}

async def main():
    user = asyncuserinfo(42)
    # 正确写法：先拿到属性（它是 coroutine），再 await
    profile1 = await user.profile
    print(profile1)
    # 第二次访问同样直接命中缓存，不再执行 async def 体
    profile2 = await user.profile
    print(profile2 is profile1)  # true

asyncio.run(main())

关键点在于 user.profile 本身返回的是一个协程对象，await 之后才得到真正的字典。propcache 内部会缓存协程执行完毕后的最终结果，因此第二次 await user.profile 时，底层不再调用原协程函数，而是直接返回已缓存的字典对象——所以 profile2 is profile1 为 true，两者指向同一内存地址。

使用限制与易错点

只能用于实例属性：装饰器依赖实例的 __dict__ 存储缓存，不能用在类级别直接访问，也不能用于 __slots__ 定义的类（因为 __slots__ 没有 __dict__）。
禁止赋值：cached_property 没有 __set__ 方法，因此 loader.raw_data = [...] 会抛出 attributeerror。如果确实需要手动清空缓存，可以删除实例属性：del loader.raw_data，下次访问会重新计算。
线程安全：propcache 的缓存写入不是原子操作。多线程同时首次访问同一实例的缓存属性时，可能重复执行方法体。如果计算有副作用或开销极大，建议外部加锁。
不要装饰普通函数：装饰器只能用于类的方法，第一个参数必须是 self。脱离类直接使用会报错。

实际项目中，cached_property 非常适合封装配置解析、正则编译、资源加载等重操作；async_cached_property 则常用于 fastapi 或 aiohttp 的请求对象中，缓存需要异步获取的用户信息、权限列表等数据，避免每个请求重复查询数据库。

3. 常用 api 详解：缓存管理、线程安全与异步支持

在进入 api 细节之前，先确保环境就绪：

pip install propcache

然后导入核心类与辅助函数：

import asyncio
from propcache import cached_property, cache_clear, cache_invalidate

3.1 缓存清理：cache_clear与cache_invalidate

propcache 的缓存并非永久驻留。当依赖的底层数据变化时，你必须主动失效缓存。最直接的方式是使用 cache_clear，它会清空某个实例上所有被 @cached_property 装饰的属性缓存。

class datafetcher:
    def __init__(self, user_id):
        self.user_id = user_id

    @cached_property
    def profile(self):
        print(f"fetching profile for {self.user_id}")
        return {"name": "alice", "level": 5}

    @cached_property
    def stats(self):
        print(f"computing stats for {self.user_id}")
        return {"wins": 10, "losses": 2}

fetcher = datafetcher(1001)
print(fetcher.profile)   # 输出 fetching... 并返回数据
print(fetcher.profile)   # 命中缓存，无输出
cache_clear(fetcher)     # 清空该实例全部缓存
print(fetcher.profile)   # 重新计算

cache_clear 接受一个实例作为参数，无返回值。它适合整体刷新场景。

如果你只想让某一个属性失效，用 cache_invalidate：

fetcher = datafetcher(1002)
_ = fetcher.profile
_ = fetcher.stats

# 只清除 profile 的缓存，stats 保留
cache_invalidate(fetcher, "profile")
print(fetcher.stats)    # 命中缓存，无打印
print(fetcher.profile)  # 重新计算

注意 cache_invalidate 的第二个参数是字符串形式的属性名。若属性不存在或未被缓存，调用不会报错，只是静默无操作。

3.2 线程安全：何时需要加锁？

propcache 底层使用原子操作保证单次读取的线程安全。在 cpython 中，多个线程同时首次访问同一个 cached_property 时，不会出现重复计算的竞态条件——库内部通过锁机制确保只有一个线程执行计算逻辑，其余线程等待结果。

import threading

class sharedresource:
    @cached_property
    def heavy_data(self):
        print("computing...")
        return sum(range(1000000))

resource = sharedresource()
threads = [threading.thread(target=lambda: resource.heavy_data) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
# 输出仅一次 "computing..."

但要注意：缓存失效操作并非线程安全。如果在多线程环境中调用 cache_clear 或 cache_invalidate，同时其他线程正在读取缓存属性，可能引发不可预知行为。建议在写操作（失效）时使用外部锁，读操作无需额外加锁。

3.3 与 asyncio 协同：异步缓存模式

propcache 的 cached_property 本身是同步的。若你的计算逻辑涉及异步 i/o（如数据库查询），需要手动包装成异步属性。推荐模式是：用 cached_property 缓存一个 asyncio.task 对象。

class asyncservice:
    def __init__(self):
        self._cache = {}

    @cached_property
    def data_task(self):
        print("creating async task...")
        return asyncio.create_task(self._load_data())

    async def _load_data(self):
        await asyncio.sleep(0.5)  # 模拟异步 i/o
        return {"status": "ready"}

async def main():
    service = asyncservice()
    # 第一次访问会启动后台任务
    task = service.data_task
    result = await task
    print(result)

    # 第二次访问直接拿到已完成的任务，不会重新创建
    result_again = await service.data_task
    print(result_again)

asyncio.run(main())

核心技巧：cached_property 缓存的是 task 对象本身，而不是结果。这样多个协程等待同一个任务时，不会重复执行异步函数。若要手动重置，调用 cache_invalidate(service, "data_task") 后，下次访问会创建新任务。

易错点：在事件循环关闭后访问 data_task 可能引发 runtimeerror，因为任务绑定了已关闭的循环。建议在服务生命周期内统一管理失效时机，避免在 finally 块中随意清缓存。

3.4 组合使用：条件失效

实际业务中常需要根据条件决定是否清缓存。例如用户更新资料后，仅使 profile 失效：

class userprofile:
    def __init__(self, uid):
        self.uid = uid
        self._version = 1

    @cached_property
    def profile(self):
        print(f"loading profile v{self._version}")
        return {"uid": self.uid, "version": self._version}

    def update_profile(self, new_data):
        self._version += 1
        # 数据版本变化，强制刷新
        cache_invalidate(self, "profile")

user = userprofile(42)
print(user.profile)   # 输出 v1
user.update_profile({"level": 99})
print(user.profile)   # 重新计算，输出 v2

此模式避免了手动跟踪脏标记，将失效逻辑封装在写操作方法内，保持调用方简洁。记住：cache_clear 作用于整个实例，而 cache_invalidate 更精确，推荐优先使用后者以减少不必要的重算开销。

4. 完整小例子：构建一个带缓存的异步数据加载器

先安装依赖：

pip install propcache aiohttp

propcache 本身不依赖 aiohttp，这里引入它只是为了模拟真实的网络请求场景。下面我们构建一个模拟异步获取用户信息的类：

import asyncio
import time
from propcache import async_cached_property
import aiohttp

class userservice:
    """模拟一个需要异步获取用户数据的服务"""
    
    def __init__(self, user_id: int):
        self.user_id = user_id
        self._fetch_count = 0  # 记录实际发起请求的次数
    
    async def _fetch_from_api(self) -> dict:
        """模拟真实的 api 调用，这里用 sleep 代替网络延迟"""
        self._fetch_count += 1
        await asyncio.sleep(1)  # 模拟 1 秒网络延迟
        
        # 模拟从远程返回的用户数据
        return {
            "id": self.user_id,
            "name": f"用户{self.user_id}",
            "email": f"user{self.user_id}@example.com",
            "level": 3,
            "points": 1000 + self.user_id * 10
        }
    
    @async_cached_property
    async def profile(self) -> dict:
        """获取用户完整资料（带缓存）"""
        return await self._fetch_from_api()
    
    @async_cached_property
    async def points(self) -> int:
        """获取用户积分（带缓存）"""
        data = await self._fetch_from_api()
        return data["points"]

async def main():
    service = userservice(42)
    
    # 第一次访问，会触发真正的数据获取
    start = time.perf_counter()
    profile = await service.profile
    first_cost = time.perf_counter() - start
    print(f"第一次访问 profile: {profile}")
    print(f"耗时: {first_cost:.2f} 秒")
    
    # 第二次访问，应该命中缓存，几乎零耗时
    start = time.perf_counter()
    profile_again = await service.profile
    second_cost = time.perf_counter() - start
    print(f"第二次访问 profile: {profile_again}")
    print(f"耗时: {second_cost:.4f} 秒")
    
    # 访问另一个缓存属性，也会触发一次请求
    start = time.perf_counter()
    pts = await service.points
    points_cost = time.perf_counter() - start
    print(f"访问 points: {pts}")
    print(f"耗时: {points_cost:.2f} 秒")
    
    # 查看总共发起了几次真实请求
    print(f"\n实际请求次数: {service._fetch_count}")
    print(f"预期请求次数: 2 (profile + points 各一次)")

asyncio.run(main())

运行这段代码，你会看到类似这样的输出：

第一次访问 profile: {'id': 42, 'name': '用户42', 'email': 'user42@example.com', 'level': 3, 'points': 1420}
耗时: 1.00 秒
第二次访问 profile: {'id': 42, 'name': '用户42', 'email': 'user42@example.com', 'level': 3, 'points': 1420}
耗时: 0.0001 秒
访问 points: 1420
耗时: 1.00 秒

实际请求次数: 2
预期请求次数: 2 (profile + points 各一次)

关键语法与易错点

1. 缓存键的粒度问题

注意 profile 和 points 是两个独立的缓存属性。虽然它们内部都调用了 _fetch_from_api()，但 propcache 会为每个属性分别缓存结果。所以上面的代码总共发起了 2 次“网络请求”，而不是 4 次。如果你希望多个属性共享同一份底层数据，建议先缓存原始数据，再派生其他属性：

class userservicebetter:
    def __init__(self, user_id: int):
        self.user_id = user_id
    
    @async_cached_property
    async def raw_data(self) -> dict:
        """只请求一次，后续所有派生属性都基于此缓存"""
        await asyncio.sleep(1)
        return {"id": self.user_id, "name": f"用户{self.user_id}", "points": 100}
    
    @async_cached_property
    async def profile(self) -> dict:
        data = await self.raw_data  # 这里会命中 raw_data 的缓存
        return {**data, "level": 3}
    
    @async_cached_property
    async def points(self) -> int:
        data = await self.raw_data  # 同样命中缓存
        return data["points"]

2. 不要手动删除缓存

propcache 的缓存默认绑定在实例上，生命周期与实例一致。如果你需要强制刷新，目前没有公开的 invalidate 方法。一个常见做法是改用普通属性 + 自行管理缓存逻辑，或者直接重建实例。

3. 并发访问安全

async_cached_property 内部使用锁机制，多个协程同时首次访问同一个属性时，只会触发一次底层调用，其余协程会等待结果并共享缓存。这在 fastapi 等并发框架中非常有用。

4. 可变返回值的陷阱

如果缓存的值是可变对象（如 list、dict），外部修改会影响后续所有读取者。上面的示例中，profile 返回的是字典，如果调用方执行 profile["name"] = "hacked"，后续所有访问都会得到被篡改的值。生产环境建议返回不可变对象或拷贝副本。

这个模式非常适合用在 fastapi 的依赖注入、爬虫会话管理、配置加载等场景，把 io 密集型的重复操作收敛到一次。

5. 进阶写法：与 fastapi 集成实现请求级缓存

fastapi 的依赖注入系统非常适合与 propcache 配合。最常见场景是：同一请求内多次调用依赖函数，但数据库查询或远程调用只应执行一次。注意这里强调“请求级”，因为 propcache 默认缓存是全局的，跨请求复用可能导致数据过期。我们通过 contextvars 在每次请求时创建独立缓存实例。

先安装依赖：

pip install fastapi propcache uvicorn

核心写法是定义一个依赖工厂，用 contextvar 保存当前请求的缓存对象：

# request_cache.py
from contextvars import contextvar
from fastapi import fastapi, depends, request
import propcache
import asyncio

# 保存当前请求的缓存实例
_cache_var: contextvar[propcache.asynccache] = contextvar(
    "request_cache", default=none
)

app = fastapi()


async def get_cache():
    """请求级缓存依赖：每个请求独立缓存"""
    cache = _cache_var.get()
    if cache is none:
        # 创建异步缓存，ttl 设为 5 秒
        cache = propcache.asynccache(ttl=5.0)
        _cache_var.set(cache)
    return cache


async def fetch_user_from_db(user_id: int):
    """模拟数据库查询，带 200ms 延迟"""
    await asyncio.sleep(0.2)
    return {"id": user_id, "name": f"user-{user_id}"}


@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    cache: propcache.asynccache = depends(get_cache),
):
    # 以 user_id 为 key 缓存查询结果
    cache_key = f"user:{user_id}"
    user = await cache.get(cache_key)
    if user is none:
        user = await fetch_user_from_db(user_id)
        await cache.set(cache_key, user)
    return {"source": "cache" if user else "db", "data": user}


@app.middleware("http")
async def reset_cache(request: request, call_next):
    """每个请求结束后重置缓存，避免跨请求污染"""
    response = await call_next(request)
    _cache_var.set(none)
    return response

关键语法解释：

contextvar 是 python 3.7+ 标准库，每个协程任务有独立上下文。fastapi 对每个请求在独立任务中处理，因此 _cache_var.set() 只在当前请求内生效。
depends(get_cache) 使缓存实例成为依赖，路由函数直接接收。若同一请求内多处 depends(get_cache)，fastapi 默认会复用同一实例，不会重复创建。
中间件在响应返回后调用 _cache_var.set(none)，清除引用，让缓存对象被垃圾回收。若省略此步，缓存实例会残留在 contextvar 中，下一个请求会误用旧数据。

一个易错点是：propcache.asynccache 的 get 方法在 key 不存在时返回 none。因此若数据库本身可能返回 none，需用 sentinel 区分。改进写法：

sentinel = object()
result = await cache.get(cache_key, default=sentinel)
if result is sentinel:
    result = await fetch_user_from_db(user_id)
    await cache.set(cache_key, result)

另外，fastapi 的异步依赖支持 yield 方式做清理，更优雅地替代中间件：

async def get_cache_yield():
    cache = propcache.asynccache(ttl=5.0)
    _cache_var.set(cache)
    try:
        yield cache
    finally:
        _cache_var.set(none)

此时路由改为 depends(get_cache_yield)，fastapi 会在请求结束时自动执行 finally 块。两种方式等价，推荐 yield 版本，逻辑更内聚。

生产环境中，数据库查询通常走连接池，耗时更长。你可以把 ttl 调整为 30~60 秒，或使用 propcache.cache（同步版）配合 def 依赖。同步依赖在 fastapi 中会跑在线程池，同样能隔离请求。记住核心原则：缓存实例的生命周期必须短于请求生命周期，否则数据会在并发请求间串扰。

6. 注意事项：避免缓存失效陷阱与内存泄漏

propcache 的缓存一旦建立，便不可变、无法手动失效。这是它的核心设计：@cached_property 装饰的属性在首次访问后，值被永久固化在实例内部，后续访问不再执行原函数。这个特性带来两个直接后果。

后果一：缓存值无法“过期”。若你的属性依赖外部状态（如数据库记录、配置文件、系统时间），那么缓存可能返回陈旧数据。例如：

from propcache import cached_property
import time

class session:
    def __init__(self, user_id):
        self.user_id = user_id

    @cached_property
    def token(self):
        # 模拟从远程服务获取 token
        return f"token-{int(time.time())}"

s = session(1)
print(s.token)  # token-1699999999
time.sleep(2)
print(s.token)  # 仍然是 token-1699999999，不会刷新

这里两次访问 s.token 得到相同结果，因为 cached_property 不会感知时间流逝。若业务要求 token 每 60 秒刷新一次，直接使用 propcache 就会出错。

后果二：无法手动清空缓存。propcache 没有提供类似 cache_clear() 或 invalidate() 的方法。你不能在运行时删除已缓存的属性。尝试 del s.token 会抛出 attributeerror。

应对策略：实例重建

最直接的解法是“换一个实例”。既然缓存绑定在实例上，那就创建新实例来获得新缓存：

class datafetcher:
    def __init__(self, source_url):
        self.source_url = source_url

    @cached_property
    def raw_data(self):
        # 模拟耗时网络请求
        return {"url": self.source_url, "payload": [1, 2, 3]}

# 第一次请求
f1 = datafetcher("https://api.example.com/v1")
print(f1.raw_data)

# 数据需要刷新时，重建实例
f2 = datafetcher("https://api.example.com/v1")
print(f2.raw_data)  # 重新执行 raw_data 函数

此模式适用于低频刷新场景，比如配置加载、模型初始化。注意：重建实例的开销必须远小于重新计算属性的开销，否则得不偿失。

版本号控制：更精细的失效

对于需要“部分失效”的场景，引入版本号字段：

import hashlib
from propcache import cached_property

class document:
    def __init__(self, content):
        self._content = content
        self._version = 1  # 手动管理的版本号

    def update_content(self, new_content):
        self._content = new_content
        self._version += 1  # 内容变更，递增版本

    @cached_property
    def checksum(self):
        return hashlib.sha256(self._content.encode()).hexdigest()

    def get_checksum(self):
        # 版本号变化时，强制重建整个对象
        if self._version != getattr(self, "_cached_version", none):
            # 删除旧缓存（通过重建副本实现）
            new_doc = document.__new__(document)
            new_doc.__dict__.update(self.__dict__)
            # 清除 cached_property 存储的缓存值
            new_doc.__dict__.pop("checksum", none)
            self.__dict__.update(new_doc.__dict__)
            self._cached_version = self._version
        return self.checksum

上面的代码展示了“伪失效”技巧：通过复制 __dict__ 并删除缓存键来重置。但注意，这已经偏离 propcache 的简洁初衷。更推荐的做法是：将版本号作为缓存键的一部分。

class documentv2:
    def __init__(self, content):
        self._content = content
        self._version = 1

    def update_content(self, new_content):
        self._content = new_content
        self._version += 1

    @cached_property
    def checksum(self):
        # 注意：这里引用了 self._version
        return (self._version, hashlib.sha256(self._content.encode()).hexdigest())

doc = documentv2("hello")
print(doc.checksum)   # (1, '2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824')
doc.update_content("world")
print(doc.checksum)   # 仍然是 (1, ...) —— 问题依旧！

这个例子暴露了关键陷阱：cached_property 只缓存一次，它不检查内部依赖是否变化。所以版本号必须在实例外部管理：

class documentv3:
    def __init__(self, content, version=1):
        self._content = content
        self.version = version  # 外部传入版本

    def with_version(self, new_version):
        # 返回新实例，版本不同则缓存自然不同
        return documentv3(self._content, new_version)

    @cached_property
    def checksum(self):
        return hashlib.sha256(self._content.encode()).hexdigest()

d1 = documentv3("hello", version=1)
print(d1.checksum)  # 2cf24dba...
d2 = d1.with_version(2)  # 新实例，缓存会重新计算
print(d2.checksum)  # 2cf24dba... 内容没变，但函数重新执行了

内存泄漏的预防

propcache 本身不会泄漏内存，但缓存的对象会持有引用。若缓存属性指向大对象（如整个 dataframe），而实例长期存活，内存就无法释放。解决方式：确保实例生命周期短，或用 weakref 包装。

import weakref
from propcache import cached_property

class heavydata:
    def __init__(self):
        self._big_list = list(range(10**6))

    @cached_property
    def summary(self):
        return sum(self._big_list)  # 缓存计算结果，而非大对象本身

# 正确用法：缓存计算结果，不要缓存大容器
h = heavydata()
print(h.summary)  # 499999500000
del h  # 实例可被垃圾回收，因为 summary 是 int，不持有大列表

若缓存属性返回 self._big_list 本身，则实例被缓存值引用，形成循环引用。propcache 不会自动打破这种环，需靠 python 垃圾回收器处理，但可能延迟释放。建议：缓存标量或小对象，不要缓存可变大容器。

实践建议

明确缓存语义：只对纯函数（相同输入必得相同输出）使用 cached_property。
依赖外部状态时：优先用实例重建或版本化新实例，而不是试图手动清缓存。
监控内存：用 tracemalloc 或 objgraph 检查是否有缓存对象意外滞留。
避免在缓存函数内捕获可变外部变量：如闭包引用的列表或字典，它们的变化不会触发缓存更新。

propcache 是“无脑缓存”工具，适合计算昂贵且结果稳定的场景。一旦需要失效机制，就该考虑换用 functools.lru_cache（支持 cache_clear()）或自行实现带 ttl 的缓存装饰器。理解这些边界，才能避免线上事故。

7. 适用场景：何时该用 propcache，何时不该用

前面六节已经把 propcache 的核心 api 和异步用法都过了一遍，最后我们冷静下来，用工程视角判断：哪些地方值得引入它，哪些地方用了反而添乱。

适合使用 propcache 的典型场景

第一类：配置读取。 程序启动时从环境变量、toml 或 yaml 文件加载配置，进程生命周期内基本不变。每次访问都重新解析文件是纯浪费，用 cached_property 懒加载并永久缓存，是最经典的做法。

import os
import tomllib
from propcache import cached_property

class appconfig:
    def __init__(self, path: str):
        self._path = path
        self._mtime: float | none = none

    @cached_property
    def settings(self) -> dict:
        """解析 toml 文件，结果缓存到实例生命周期结束。"""
        with open(self._path, "rb") as f:
            return tomllib.load(f)

    @cached_property
    def debug_enabled(self) -> bool:
        """环境变量读取后缓存，避免每次访问都触发 os.environ 查询。"""
        return os.getenv("app_debug", "0") == "1"

cfg = appconfig("app.toml")
print(cfg.settings)   # 第一次解析文件
print(cfg.settings)   # 直接命中缓存，不再读磁盘

注意 settings 返回的是可变 dict，外部拿到引用后可以修改内部值，但 propcache 不会重新计算。如果你需要防止外部篡改，返回 mappingproxytype 或冻结副本。

第二类：重复计算。 某个属性需要经过排序、聚合或正则匹配等 cpu 密集操作，且输入在实例创建后不变。典型例子是数据校验规则预编译：

import re
from propcache import cached_property

class regexvalidator:
    def __init__(self, pattern: str):
        self.pattern = pattern

    @cached_property
    def _compiled(self) -> re.pattern:
        return re.compile(self.pattern)

    def is_valid(self, text: str) -> bool:
        return self._compiled.fullmatch(text) is not none

v = regexvalidator(r"[a-z]+@[a-z]+\.[a-z]{2,3}")
for _ in range(1000):
    v.is_valid("user@example.com")   # 编译只发生一次

re.compile 在 cpython 内部也有缓存，但容量有限；当你有大量不同正则实例时，cached_property 能精确控制每个实例的编译结果，不依赖全局 lru。

第三类：异步 i/o 结果缓存。 从数据库或 http api 拉取不常变化的数据（如用户角色表、国家代码列表），用 async_cached_property 避免重复网络请求：

import asyncio
import aiohttp
from propcache import async_cached_property

class remotecatalog:
    def __init__(self, base_url: str):
        self._base = base_url

    @async_cached_property
    async def categories(self) -> list[str]:
        async with aiohttp.clientsession() as session:
            async with session.get(f"{self._base}/api/categories") as resp:
                resp.raise_for_status()
                return await resp.json()

async def main():
    cat = remotecatalog("https://example.com")
    first = await cat.categories    # 触发 http 请求
    second = await cat.categories   # 返回缓存，无网络调用
    assert first is second          # 同一对象

asyncio.run(main())

关键点：async_cached_property 在多个协程并发首次访问时，内部通过 asyncio.lock 保证只发一次请求，其余协程等待同一个结果。这是手写 _cache = none + if 判断最容易出错的地方。

不适合使用的场景

频繁变化的数据绝对不要用 propcache。 比如股票实时行情、websocket 推送的传感器读数、每秒钟都会更新的任务进度——这些数据天然需要每次访问都取最新值。cached_property 一旦缓存就永远不失效（除非手动 del obj.attr），你会在不知情中读到几分钟前的旧数据，而且很难排查。

实例生命周期极短的对象也不值得用。 如果一个对象只存活几毫秒就被销毁，缓存属性反而多一次 __dict__ 写入和内存分配，得不偿失。propcache 的底层原理是把值写入实例的 __dict__，与普通实例属性相比没有魔法，只是多了“只计算一次”的语义。对于一次性临时对象，直接用普通属性或局部变量即可。

另一个常见误区：不要用 cached_property 包装依赖外部可变状态的方法。 例如属性值依赖某个全局配置项或数据库连接状态，而这些状态可能在运行时被修改。propcache 不追踪依赖关系，缓存后不会自动失效。

from propcache import cached_property

class worker:
    def __init__(self):
        self.threshold = 10

    @cached_property
    def is_ready(self) -> bool:
        return self.threshold > 5   # 危险：threshold 可变但结果被缓存

w = worker()
print(w.is_ready)   # true
w.threshold = 1     # 修改后 is_ready 仍是 true（缓存）

如果你需要“依赖变化时自动失效”，应当改用 cached_property 配合手动删除属性，或者直接使用 functools.cached_property 的变体 + 事件通知机制。propcache 定位是“确定性输入下的确定性缓存”，不是通用响应式缓存。

最后一条实用建议：用 cached_property 时，属性名不要以下划线开头（如 _data），否则 propcache 会把它当作私有属性，缓存键冲突且语义混乱。保持公开命名，用 del obj.attr 显式重置即可。

安装和完整 api 清单见前文，这里不再重复。记住一句话：“低频计算 + 高频读取 + 输入稳定” 是 propcache 的黄金三角，三者缺一就要重新评估。

以上就是python使用propcache缓存属性与异步接口的详细内容，更多关于python propcache缓存属性与异步接口的资料请关注代码网其它相关文章！

Python使用propcache缓存属性与异步接口

2026年09月15日 • Python •我要评论