Python 类型提示：Callable 后跟 TypeVar 是什么意思？

Question

我试图理解Getter[T]以下代码中的类型提示：

简化示例

T = TypeVar('T')
Getter = Callable[[T, str], str]


class AbstractClass(abc.ABC):
    @abc.abstractmethod
    def extract(
        self,
        get_from_carrier: Getter[T],  #  <---- See here
        ...
    ) -> Context:

非常感谢您的帮助，因为我一直为此而伤透脑筋。

原始源代码

原始源代码来自OpenTelemetry 项目文件“textmap.py”：

import abc
import typing

from opentelemetry.context.context import Context

TextMapPropagatorT = typing.TypeVar("TextMapPropagatorT")

Setter = typing.Callable[[TextMapPropagatorT, str, str], None]
Getter = typing.Callable[[TextMapPropagatorT, str], typing.List[str]]


class TextMapPropagator(abc.ABC):
    """This class provides an interface that enables extracting and injecting
    context into headers of HTTP requests. 
    ...
    """

    @abc.abstractmethod
    def extract(
        self,
        get_from_carrier: Getter[TextMapPropagatorT],
        carrier: TextMapPropagatorT,
        context: typing.Optional[Context] = None,
    ) -> Context:

Answer 1

Callable 后跟一个类型变量意味着该 callable 是一个泛型函数，它接受一个或多个泛型类型的参数T。

类型变量T是任何泛型类型的参数。

该行：

Getter = Callable[[T, str], str]

定义Getter为可调用函数的类型别名，该函数的参数为​​泛型类型T和字符串，返回类型为字符串。

因此，该行：

get_from_carrier: Getter[T]

定义一个参数 ( get_from_carrier)，它是一个泛型函数。泛型函数的第一个参数是泛型类型T。

具体例子

通过看一个具体的例子可以更好地理解这一点。请参阅propagators.extract下面的“instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/ init .py ”：

在调用中propagators.extract，该函数get_header_from_scope是一个可调用函数，其第一个参数的类型为dict，这dict充当TextMapPropagatorT.

def get_header_from_scope(scope: dict, header_name: str) -> typing.List[str]:
    """Retrieve a HTTP header value from the ASGI scope.

    Returns:
        A list with a single string with the header value if it exists, else an empty list.
    """
    headers = scope.get("headers")
    return [
        value.decode("utf8")
        for (key, value) in headers
        if key.decode("utf8") == header_name
    ]


...


class OpenTelemetryMiddleware:
    """The ASGI application middleware.
    ...
    """

    ...

    async def __call__(self, scope, receive, send):
        """The ASGI application  ... """
        if scope["type"] not in ("http", "websocket"):
            return await self.app(scope, receive, send)

        token = context.attach(
            propagators.extract(get_header_from_scope, scope)
        )