Python 类型提示 : What does Callable followed by a TypeVar mean?

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 project file "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。

具体例子

通过查看具体示例可以更好地理解这一点。请参阅下面来自 "instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/init.py " 的 propagators.extract :

在调用 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)
        )