Python异步场景下如何正确使用returns库的RequiresContext相关类？

核心函数修正 #

from redis.asyncio import Redis
from returns.context import RequiresContextFutureResultE
from returns.result import ResultE, safe


def get_value(key: str) -> RequiresContextFutureResultE[str, Redis]:
    @safe
    def decode_value(value: bytes | None) -> str:
        return value.decode("utf-8")

    async def inner(client: Redis) -> ResultE[str]:
        raw_value = await client.get(key)
        return decode_value(raw_value)

    return RequiresContextFutureResultE(inner)

运行代码 #

import asyncio

async def main():
    async with Redis(
        host="localhost",
        port=6379,
        password="902nks291",
        db=15,
    ) as redis_client:
        rcfr = get_value(key="user-id")
        value_result = await rcfr(redis_client)
        print("value: ", value_result.unwrap())

asyncio.run(main())

关键修正说明 #

• 移除get_value的async修饰符：RequiresContextFutureResultE是同步构造的上下文容器，无需将整个函数声明为异步。
• 异常自动捕获：用@safe装饰解码逻辑，自动处理None.decode()这类潜在异常，返回ResultE类型统一处理成功/失败场景。
• 简化异步逻辑：内部异步函数直接返回ResultE即可，无需手动包装FutureResult，RequiresContextFutureResultE会自动适配异步返回值。
• 明确类型边界：导入redis.asyncio.Redis而非同步版，避免类型混淆。
• 结果处理：rcfr(redis_client)返回可等待的ResultE[str]，await后通过unwrap()获取实际值（也可使用match/map等方法优雅处理异常分支）。

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