主页 / user-8166878

Alice Yu's questions

Martin Hope
Alice Yu
Asked: 2025-04-09 03:12:21 +0800 CST
  • 6

我有一个现有的 Python 装饰器，它确保方法被赋予一个 psycopg AsyncConnection 实例。我正在尝试更新要使用的类型ParamSpec，Concatenate因为当前的实现不是类型安全的，但我遇到了问题。

当前的实现如下：

def ensure_conn(func: Callable[..., Coroutine[Any, Any, R]]) -> Callable[..., Coroutine[Any, Any, R]]:
    """Ensure the function has a conn argument. If conn is not provided, generate a new connection and pass it to the function."""

    async def wrapper(*args: Any, **kwargs: Any) -> R:
        # Get named keyword argument conn, or find an AsyncConnection in the args
        kwargs_conn = kwargs.get("conn")
        conn_arg: AsyncConnection[Any] | None = None
        if isinstance(kwargs_conn, AsyncConnection):
            conn_arg = kwargs_conn
        elif not conn_arg:
            for arg in args:
                if isinstance(arg, AsyncConnection):
                    conn_arg = arg
                    break
        if conn_arg:
            # If conn is provided, call the method as is
            return await func(*args, **kwargs)
        else:
            # If conn is not provided, generate a new connection and pass it to the method
            db_driver = DbDriver()
            async with db_driver.connection() as conn:
                return await func(*args, **kwargs, conn=conn)

    return wrapper

当前使用情况：

@ensure_conn
async def get_user(user_id: UUID, conn: AsyncConnection):
    async with conn.cursor() as cursor:
    // do stuff

...但我可以这样调用它，并且它不会失败类型检查：

get_user('519766c5-af86-47ea-9fa9-cee0c0de66b1', conn, arg_that_should_fail_typing)

以下是我目前使用ParamSpec和 所得到的最接近的实现Concatenate：

def ensure_conn_decorator[**P, R](func: Callable[Concatenate[AsyncConnection[Any], P], R]) -> Coroutine[Any, Any, R]:
    """Ensure the function has a conn argument. If conn is not provided, generate a new connection and pass it to the function."""
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        # Get named keyword argument conn, or find an AsyncConnection in the args
        kwargs_conn = kwargs.get("conn")
        conn_arg: AsyncConnection[Any] | None = None
        if isinstance(kwargs_conn, AsyncConnection):
            conn_arg = kwargs_conn
        elif not conn_arg:
            for arg in args:
                if isinstance(arg, AsyncConnection):
                    conn_arg = arg
                    break
        if conn_arg:
            # If conn is provided, call the method as is
            return await func(*args, **kwargs)
        else:
            # If conn is not provided, generate a new connection and pass it to the method
            db_driver = DbDriver()
            async with db_driver.connection() as conn:
                return await func(*args, **kwargs, conn=conn)

    return wrapper

问题是

  • Conn 必须是第一个方法参数，而不是任意位置 - 它通常是任意 X 个参数之后的最后一个参数
  • 无法确定正确的返回类型
Expression of type "(**P@ensure_conn_decorator) -> Coroutine[Any, Any, R@ensure_conn_decorator]" is incompatible with return type "Coroutine[Any, Any, R@ensure_conn_decorator]"
  "function" is incompatible with "Coroutine[Any, Any, R@ensure_conn_decorator]"
python