Make email.policy classes generic (#12724)

stdlib/@tests/stubtest_allowlists/common.txt

@@ -360,6 +360,8 @@ codecs.CodecInfo.incrementalencoder
 codecs.CodecInfo.streamreader
 codecs.CodecInfo.streamwriter
 
+contextvars.Context.__init__  # C signature is broader than what is actually accepted
+
 # The Dialect properties are initialized as None in Dialect but their values are enforced in _Dialect
 csv.Dialect.delimiter
 csv.Dialect.doublequote
@@ -370,8 +372,8 @@ csv.Dialect.skipinitialspace
 csv.DictReader.__init__  # runtime sig has *args but will error if more than 5 positional args are supplied
 csv.DictWriter.__init__  # runtime sig has *args but will error if more than 5 positional args are supplied
 
-contextvars.Context.__init__  # C signature is broader than what is actually accepted
 dataclasses.field  # White lies around defaults
+email.policy.EmailPolicy.message_factory  # "type" at runtime, but protocol in stubs
 hashlib.scrypt  # Raises TypeError if salt, n, r or p are None
 hashlib.sha3_\d+  # Can be a class or a built-in function, can't be subclassed at runtime
 hashlib.shake_\d+  # Can be a class or a built-in function, can't be subclassed at runtime

stdlib/email/_policybase.pyi

@@ -1,11 +1,29 @@
 from abc import ABCMeta, abstractmethod
-from collections.abc import Callable
 from email.errors import MessageDefect
 from email.header import Header
 from email.message import Message
+from typing import Generic, Protocol, TypeVar, type_check_only
 from typing_extensions import Self
 
-class _PolicyBase:
+_MessageT = TypeVar("_MessageT", bound=Message, default=Message)
+
+@type_check_only
+class _MessageFactory(Protocol[_MessageT]):
+    def __call__(self, policy: Policy[_MessageT]) -> _MessageT: ...
+
+# Policy below is the only known direct subclass of _PolicyBase. We therefore
+# assume that the __init__ arguments and attributes of _PolicyBase are
+# the same as those of Policy.
+class _PolicyBase(Generic[_MessageT]):
+    max_line_length: int | None
+    linesep: str
+    cte_type: str
+    raise_on_defect: bool
+    mangle_from_: bool
+    message_factory: _MessageFactory[_MessageT] | None
+    # Added in Python 3.8.20, 3.9.20, 3.10.15, 3.11.10, 3.12.5
+    verify_generated_headers: bool
+
     def __init__(
         self,
         *,
@@ -14,7 +32,7 @@ class _PolicyBase:
         cte_type: str = "8bit",
         raise_on_defect: bool = False,
         mangle_from_: bool = ...,  # default depends on sub-class
-        message_factory: Callable[[Policy], Message] | None = None,
+        message_factory: _MessageFactory[_MessageT] | None = None,
         # Added in Python 3.8.20, 3.9.20, 3.10.15, 3.11.10, 3.12.5
         verify_generated_headers: bool = True,
     ) -> None: ...
@@ -26,24 +44,15 @@ class _PolicyBase:
         cte_type: str = ...,
         raise_on_defect: bool = ...,
         mangle_from_: bool = ...,
-        message_factory: Callable[[Policy], Message] | None = ...,
+        message_factory: _MessageFactory[_MessageT] | None = ...,
         # Added in Python 3.8.20, 3.9.20, 3.10.15, 3.11.10, 3.12.5
         verify_generated_headers: bool = ...,
     ) -> Self: ...
     def __add__(self, other: Policy) -> Self: ...
 
-class Policy(_PolicyBase, metaclass=ABCMeta):
-    max_line_length: int | None
-    linesep: str
-    cte_type: str
-    raise_on_defect: bool
-    mangle_from_: bool
-    message_factory: Callable[[Policy], Message] | None
-    # Added in Python 3.8.20, 3.9.20, 3.10.15, 3.11.10, 3.12.5
-    verify_generated_headers: bool
-
-    def handle_defect(self, obj: Message, defect: MessageDefect) -> None: ...
-    def register_defect(self, obj: Message, defect: MessageDefect) -> None: ...
+class Policy(_PolicyBase[_MessageT], metaclass=ABCMeta):
+    def handle_defect(self, obj: _MessageT, defect: MessageDefect) -> None: ...
+    def register_defect(self, obj: _MessageT, defect: MessageDefect) -> None: ...
     def header_max_count(self, name: str) -> int | None: ...
     @abstractmethod
     def header_source_parse(self, sourcelines: list[str]) -> tuple[str, str]: ...
@@ -56,11 +65,11 @@ class Policy(_PolicyBase, metaclass=ABCMeta):
     @abstractmethod
     def fold_binary(self, name: str, value: str) -> bytes: ...
 
-class Compat32(Policy):
+class Compat32(Policy[_MessageT]):
     def header_source_parse(self, sourcelines: list[str]) -> tuple[str, str]: ...
     def header_store_parse(self, name: str, value: str) -> tuple[str, str]: ...
     def header_fetch_parse(self, name: str, value: str) -> str | Header: ...  # type: ignore[override]
     def fold(self, name: str, value: str) -> str: ...
     def fold_binary(self, name: str, value: str) -> bytes: ...
 
-compat32: Compat32
+compat32: Compat32[Message]

stdlib/email/feedparser.pyi

@@ -5,19 +5,19 @@ from typing import Generic, TypeVar, overload
 
 __all__ = ["FeedParser", "BytesFeedParser"]
 
-_MessageT = TypeVar("_MessageT", bound=Message)
+_MessageT = TypeVar("_MessageT", bound=Message, default=Message)
 
 class FeedParser(Generic[_MessageT]):
     @overload
-    def __init__(self: FeedParser[Message], _factory: None = None, *, policy: Policy = ...) -> None: ...
+    def __init__(self: FeedParser[Message], _factory: None = None, *, policy: Policy[Message] = ...) -> None: ...
     @overload
-    def __init__(self, _factory: Callable[[], _MessageT], *, policy: Policy = ...) -> None: ...
+    def __init__(self, _factory: Callable[[], _MessageT], *, policy: Policy[_MessageT] = ...) -> None: ...
     def feed(self, data: str) -> None: ...
     def close(self) -> _MessageT: ...
 
 class BytesFeedParser(FeedParser[_MessageT]):
     @overload
-    def __init__(self: BytesFeedParser[Message], _factory: None = None, *, policy: Policy = ...) -> None: ...
+    def __init__(self: BytesFeedParser[Message], _factory: None = None, *, policy: Policy[Message] = ...) -> None: ...
     @overload
-    def __init__(self, _factory: Callable[[], _MessageT], *, policy: Policy = ...) -> None: ...
+    def __init__(self, _factory: Callable[[], _MessageT], *, policy: Policy[_MessageT] = ...) -> None: ...
     def feed(self, data: bytes | bytearray) -> None: ...  # type: ignore[override]

stdlib/email/generator.pyi

@@ -1,34 +1,71 @@
 from _typeshed import SupportsWrite
 from email.message import Message
 from email.policy import Policy
+from typing import Any, Generic, TypeVar, overload
 from typing_extensions import Self
 
 __all__ = ["Generator", "DecodedGenerator", "BytesGenerator"]
 
-class Generator:
-    def clone(self, fp: SupportsWrite[str]) -> Self: ...
-    def write(self, s: str) -> None: ...
+# By default, generators do not have a message policy.
+_MessageT = TypeVar("_MessageT", bound=Message, default=Any)
+
+class Generator(Generic[_MessageT]):
+    maxheaderlen: int | None
+    policy: Policy[_MessageT] | None
+    @overload
+    def __init__(
+        self: Generator[Any],  # The Policy of the message is used.
+        outfp: SupportsWrite[str],
+        mangle_from_: bool | None = None,
+        maxheaderlen: int | None = None,
+        *,
+        policy: None = None,
+    ) -> None: ...
+    @overload
     def __init__(
         self,
         outfp: SupportsWrite[str],
         mangle_from_: bool | None = None,
         maxheaderlen: int | None = None,
         *,
-        policy: Policy | None = None,
+        policy: Policy[_MessageT],
     ) -> None: ...
-    def flatten(self, msg: Message, unixfrom: bool = False, linesep: str | None = None) -> None: ...
+    def write(self, s: str) -> None: ...
+    def flatten(self, msg: _MessageT, unixfrom: bool = False, linesep: str | None = None) -> None: ...
+    def clone(self, fp: SupportsWrite[str]) -> Self: ...
 
-class BytesGenerator(Generator):
+class BytesGenerator(Generator[_MessageT]):
+    @overload
+    def __init__(
+        self: BytesGenerator[Any],  # The Policy of the message is used.
+        outfp: SupportsWrite[bytes],
+        mangle_from_: bool | None = None,
+        maxheaderlen: int | None = None,
+        *,
+        policy: None = None,
+    ) -> None: ...
+    @overload
     def __init__(
         self,
         outfp: SupportsWrite[bytes],
         mangle_from_: bool | None = None,
         maxheaderlen: int | None = None,
         *,
-        policy: Policy | None = None,
+        policy: Policy[_MessageT],
     ) -> None: ...
 
-class DecodedGenerator(Generator):
+class DecodedGenerator(Generator[_MessageT]):
+    @overload
+    def __init__(
+        self: DecodedGenerator[Any],  # The Policy of the message is used.
+        outfp: SupportsWrite[str],
+        mangle_from_: bool | None = None,
+        maxheaderlen: int | None = None,
+        fmt: str | None = None,
+        *,
+        policy: None = None,
+    ) -> None: ...
+    @overload
     def __init__(
         self,
         outfp: SupportsWrite[str],
@@ -36,5 +73,5 @@ class DecodedGenerator(Generator):
         maxheaderlen: int | None = None,
         fmt: str | None = None,
         *,
-        policy: Policy | None = None,
+        policy: Policy[_MessageT],
     ) -> None: ...

stdlib/email/message.pyi

@@ -30,10 +30,13 @@ class _SupportsDecodeToPayload(Protocol):
     def decode(self, encoding: str, errors: str, /) -> _PayloadType | _MultipartPayloadType: ...
 
 class Message(Generic[_HeaderT, _HeaderParamT]):
-    policy: Policy  # undocumented
+    # The policy attributes and arguments in this class and its subclasses
+    # would ideally use Policy[Self], but this is not possible.
+    policy: Policy[Any]  # undocumented
     preamble: str | None
     epilogue: str | None
     defects: list[MessageDefect]
+    def __init__(self, policy: Policy[Any] = ...) -> None: ...
     def is_multipart(self) -> bool: ...
     def set_unixfrom(self, unixfrom: str) -> None: ...
     def get_unixfrom(self) -> str | None: ...
@@ -126,8 +129,8 @@ class Message(Generic[_HeaderT, _HeaderParamT]):
     def get_charsets(self, failobj: _T) -> list[str | _T]: ...
     def walk(self) -> Generator[Self, None, None]: ...
     def get_content_disposition(self) -> str | None: ...
-    def as_string(self, unixfrom: bool = False, maxheaderlen: int = 0, policy: Policy | None = None) -> str: ...
-    def as_bytes(self, unixfrom: bool = False, policy: Policy | None = None) -> bytes: ...
+    def as_string(self, unixfrom: bool = False, maxheaderlen: int = 0, policy: Policy[Any] | None = None) -> str: ...
+    def as_bytes(self, unixfrom: bool = False, policy: Policy[Any] | None = None) -> bytes: ...
     def __bytes__(self) -> bytes: ...
     def set_param(
         self,
@@ -139,13 +142,12 @@ class Message(Generic[_HeaderT, _HeaderParamT]):
         language: str = "",
         replace: bool = False,
     ) -> None: ...
-    def __init__(self, policy: Policy = ...) -> None: ...
     # The following two methods are undocumented, but a source code comment states that they are public API
     def set_raw(self, name: str, value: _HeaderParamT) -> None: ...
     def raw_items(self) -> Iterator[tuple[str, _HeaderT]]: ...
 
 class MIMEPart(Message[_HeaderRegistryT, _HeaderRegistryParamT]):
-    def __init__(self, policy: Policy | None = None) -> None: ...
+    def __init__(self, policy: Policy[Any] | None = None) -> None: ...
     def get_body(self, preferencelist: Sequence[str] = ("related", "html", "plain")) -> MIMEPart[_HeaderRegistryT] | None: ...
     def attach(self, payload: Self) -> None: ...  # type: ignore[override]
     # The attachments are created via type(self) in the attach method. It's theoretically
@@ -163,7 +165,7 @@ class MIMEPart(Message[_HeaderRegistryT, _HeaderRegistryParamT]):
     def add_attachment(self, *args: Any, content_manager: ContentManager | None = ..., **kw: Any) -> None: ...
     def clear(self) -> None: ...
     def clear_content(self) -> None: ...
-    def as_string(self, unixfrom: bool = False, maxheaderlen: int | None = None, policy: Policy | None = None) -> str: ...
+    def as_string(self, unixfrom: bool = False, maxheaderlen: int | None = None, policy: Policy[Any] | None = None) -> str: ...
     def is_attachment(self) -> bool: ...
 
 class EmailMessage(MIMEPart): ...

stdlib/email/parser.pyi

@@ -12,9 +12,9 @@ _MessageT = TypeVar("_MessageT", bound=Message, default=Message)
 
 class Parser(Generic[_MessageT]):
     @overload
-    def __init__(self: Parser[Message[str, str]], _class: None = None, *, policy: Policy = ...) -> None: ...
+    def __init__(self: Parser[Message[str, str]], _class: None = None, *, policy: Policy[Message[str, str]] = ...) -> None: ...
     @overload
-    def __init__(self, _class: Callable[[], _MessageT], *, policy: Policy = ...) -> None: ...
+    def __init__(self, _class: Callable[[], _MessageT], *, policy: Policy[_MessageT] = ...) -> None: ...
     def parse(self, fp: SupportsRead[str], headersonly: bool = False) -> _MessageT: ...
     def parsestr(self, text: str, headersonly: bool = False) -> _MessageT: ...
 
@@ -25,9 +25,11 @@ class HeaderParser(Parser[_MessageT]):
 class BytesParser(Generic[_MessageT]):
     parser: Parser[_MessageT]
     @overload
-    def __init__(self: BytesParser[Message[str, str]], _class: None = None, *, policy: Policy = ...) -> None: ...
+    def __init__(
+        self: BytesParser[Message[str, str]], _class: None = None, *, policy: Policy[Message[str, str]] = ...
+    ) -> None: ...
     @overload
-    def __init__(self, _class: Callable[[], _MessageT], *, policy: Policy = ...) -> None: ...
+    def __init__(self, _class: Callable[[], _MessageT], *, policy: Policy[_MessageT] = ...) -> None: ...
     def parse(self, fp: _WrappedBuffer, headersonly: bool = False) -> _MessageT: ...
     def parsebytes(self, text: bytes | bytearray, headersonly: bool = False) -> _MessageT: ...
 

stdlib/email/policy.pyi

@@ -1,16 +1,34 @@
 from collections.abc import Callable
-from email._policybase import Compat32 as Compat32, Policy as Policy, compat32 as compat32
+from email._policybase import Compat32 as Compat32, Policy as Policy, _MessageFactory, compat32 as compat32
 from email.contentmanager import ContentManager
-from email.message import Message
-from typing import Any
+from email.message import EmailMessage, Message
+from typing import Any, TypeVar, overload
 
 __all__ = ["Compat32", "compat32", "Policy", "EmailPolicy", "default", "strict", "SMTP", "HTTP"]
 
-class EmailPolicy(Policy):
+_MessageT = TypeVar("_MessageT", bound=Message, default=Message)
+
+class EmailPolicy(Policy[_MessageT]):
     utf8: bool
     refold_source: str
     header_factory: Callable[[str, Any], Any]
     content_manager: ContentManager
+    @overload
+    def __init__(
+        self: EmailPolicy[EmailMessage],
+        *,
+        max_line_length: int | None = ...,
+        linesep: str = ...,
+        cte_type: str = ...,
+        raise_on_defect: bool = ...,
+        mangle_from_: bool = ...,
+        message_factory: None = None,
+        utf8: bool = ...,
+        refold_source: str = ...,
+        header_factory: Callable[[str, str], str] = ...,
+        content_manager: ContentManager = ...,
+    ) -> None: ...
+    @overload
     def __init__(
         self,
         *,
@@ -19,7 +37,7 @@ class EmailPolicy(Policy):
         cte_type: str = ...,
         raise_on_defect: bool = ...,
         mangle_from_: bool = ...,
-        message_factory: Callable[[Policy], Message] | None = ...,
+        message_factory: _MessageFactory[_MessageT] | None = ...,
         utf8: bool = ...,
         refold_source: str = ...,
         header_factory: Callable[[str, str], str] = ...,
@@ -31,8 +49,8 @@ class EmailPolicy(Policy):
     def fold(self, name: str, value: str) -> Any: ...
     def fold_binary(self, name: str, value: str) -> bytes: ...
 
-default: EmailPolicy
-SMTP: EmailPolicy
-SMTPUTF8: EmailPolicy
-HTTP: EmailPolicy
-strict: EmailPolicy
+default: EmailPolicy[EmailMessage]
+SMTP: EmailPolicy[EmailMessage]
+SMTPUTF8: EmailPolicy[EmailMessage]
+HTTP: EmailPolicy[EmailMessage]
+strict: EmailPolicy[EmailMessage]