Various improvements for the mailbox stubs

* Use `_typeshed.SupportsItems` instead of custom `_HasItems` protocol.
* Use a custom covariant protocol for mailbox messages.
* Add `Mailbox._dump_message`.
* Return a protocol from abstract method `Mailbox.get_file()`.
* Return concrete type `BytesIO` instead of `typing.IO` from
  `Babyl.get_file()`.
* Use our custom protocol instead of semi-protocol `typing.IO`
  in the remaining cases.

Closes: #14935

Author: srittau

stdlib/@tests/test_cases/check_mailbox.py

@@ -0,0 +1,11 @@
+import mailbox
+
+
+def mbox1() -> mailbox.Mailbox:
+    return mailbox.mbox("")
+
+def mbox2() -> mailbox.Mailbox[mailbox.mboxMessage]:
+    return mailbox.mbox("")
+
+def mbox3() -> mailbox.Mailbox[mailbox.Message]:
+    return mailbox.mbox("")

stdlib/mailbox.pyi

@@ -1,12 +1,11 @@
 import email.message
 import io
 import sys
-from _typeshed import StrPath, SupportsNoArgReadline, SupportsRead
+from _typeshed import StrPath, SupportsItems, SupportsNoArgReadline, SupportsRead, SupportsWrite, Unused
 from abc import ABCMeta, abstractmethod
 from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
-from email._policybase import _MessageT
 from types import GenericAlias, TracebackType
-from typing import IO, Any, AnyStr, Generic, Literal, Protocol, TypeVar, overload, type_check_only
+from typing import Any, AnyStr, Generic, Literal, Protocol, TypeVar, overload, type_check_only
 from typing_extensions import Self, TypeAlias
 
 __all__ = [
@@ -30,27 +29,50 @@ __all__ = [
 ]
 
 _T = TypeVar("_T")
+_AnyStr = TypeVar("_AnyStr", str, bytes, default=bytes)
 
 @type_check_only
 class _SupportsReadAndReadline(SupportsRead[bytes], SupportsNoArgReadline[bytes], Protocol): ...
 
+# As opposed to _MessageT_co in email._policybase, this type is bound to
+# mailbox.Message instead of email.message.Message.
+_MessageT_co = TypeVar("_MessageT_co", bound=Message, default=Message, covariant=True)
+
 _MessageData: TypeAlias = email.message.Message | bytes | str | io.StringIO | _SupportsReadAndReadline
 
 @type_check_only
 class _HasIteritems(Protocol):
     def iteritems(self) -> Iterator[tuple[str, _MessageData]]: ...
 
-@type_check_only
-class _HasItems(Protocol):
-    def items(self) -> Iterator[tuple[str, _MessageData]]: ...
-
 linesep: bytes
 
-class Mailbox(Generic[_MessageT]):
+# Common interface for get_file() return types.
+@type_check_only
+class _GetFileReturn(Protocol[_AnyStr]):
+    def __iter__(self) -> Iterator[_AnyStr]: ...
+    def __enter__(self) -> Self: ...
+    def __exit__(
+        self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None, /
+    ) -> bool | None: ...
+    def read(self, size: int | None = None, /) -> _AnyStr: ...
+    def read1(self, size: int | None = None, /) -> _AnyStr: ...
+    def readline(self, size: int | None = None, /) -> _AnyStr: ...
+    def readlines(self, sizehint: int | None = None, /) -> list[_AnyStr]: ...
+    def tell(self) -> int: ...
+    def seek(self, offset: int, whence: int = 0, /) -> object: ...
+    def close(self) -> object: ...
+    def readable(self) -> bool: ...
+    def writable(self) -> bool: ...
+    def seekable(self) -> bool: ...
+    def flush(self) -> object: ...
+    @property
+    def closed(self) -> bool: ...
+
+class Mailbox(Generic[_MessageT_co]):
     _path: str  # undocumented
-    _factory: Callable[[IO[Any]], _MessageT] | None  # undocumented
+    _factory: Callable[[_GetFileReturn], _MessageT_co] | None  # undocumented
     @overload
-    def __init__(self, path: StrPath, factory: Callable[[IO[Any]], _MessageT], create: bool = True) -> None: ...
+    def __init__(self, path: StrPath, factory: Callable[[_GetFileReturn], _MessageT_co], create: bool = True) -> None: ...
     @overload
     def __init__(self, path: StrPath, factory: None = None, create: bool = True) -> None: ...
     @abstractmethod
@@ -62,37 +84,38 @@ class Mailbox(Generic[_MessageT]):
     @abstractmethod
     def __setitem__(self, key: str, message: _MessageData) -> None: ...
     @overload
-    def get(self, key: str, default: None = None) -> _MessageT | None: ...
+    def get(self, key: str, default: None = None) -> _MessageT_co | None: ...
     @overload
-    def get(self, key: str, default: _T) -> _MessageT | _T: ...
-    def __getitem__(self, key: str) -> _MessageT: ...
+    def get(self, key: str, default: _T) -> _MessageT_co | _T: ...
+    def __getitem__(self, key: str) -> _MessageT_co: ...
     @abstractmethod
-    def get_message(self, key: str) -> _MessageT: ...
+    def get_message(self, key: str) -> _MessageT_co: ...
     def get_string(self, key: str) -> str: ...
     @abstractmethod
     def get_bytes(self, key: str) -> bytes: ...
-    # As '_ProxyFile' doesn't implement the full IO spec, and BytesIO is incompatible with it, get_file return is Any here
     @abstractmethod
-    def get_file(self, key: str) -> Any: ...
+    def get_file(self, key: str) -> _GetFileReturn: ...
     @abstractmethod
     def iterkeys(self) -> Iterator[str]: ...
     def keys(self) -> list[str]: ...
-    def itervalues(self) -> Iterator[_MessageT]: ...
-    def __iter__(self) -> Iterator[_MessageT]: ...
-    def values(self) -> list[_MessageT]: ...
-    def iteritems(self) -> Iterator[tuple[str, _MessageT]]: ...
-    def items(self) -> list[tuple[str, _MessageT]]: ...
+    def itervalues(self) -> Iterator[_MessageT_co]: ...
+    def __iter__(self) -> Iterator[_MessageT_co]: ...
+    def values(self) -> list[_MessageT_co]: ...
+    def iteritems(self) -> Iterator[tuple[str, _MessageT_co]]: ...
+    def items(self) -> list[tuple[str, _MessageT_co]]: ...
     @abstractmethod
     def __contains__(self, key: str) -> bool: ...
     @abstractmethod
     def __len__(self) -> int: ...
     def clear(self) -> None: ...
     @overload
-    def pop(self, key: str, default: None = None) -> _MessageT | None: ...
+    def pop(self, key: str, default: None = None) -> _MessageT_co | None: ...
     @overload
-    def pop(self, key: str, default: _T) -> _MessageT | _T: ...
-    def popitem(self) -> tuple[str, _MessageT]: ...
-    def update(self, arg: _HasIteritems | _HasItems | Iterable[tuple[str, _MessageData]] | None = None) -> None: ...
+    def pop(self, key: str, default: _T) -> _MessageT_co | _T: ...
+    def popitem(self) -> tuple[str, _MessageT_co]: ...
+    def update(
+        self, arg: _HasIteritems | SupportsItems[str, _MessageData] | Iterable[tuple[str, _MessageData]] | None = None
+    ) -> None: ...
     @abstractmethod
     def flush(self) -> None: ...
     @abstractmethod
@@ -101,16 +124,18 @@ class Mailbox(Generic[_MessageT]):
     def unlock(self) -> None: ...
     @abstractmethod
     def close(self) -> None: ...
+    # Undocumented, called by subclasses to parse added messages.
+    def _dump_message(self, message: _MessageData, target: SupportsWrite[bytes], mangle_from_: bool = False) -> None: ...
     def __class_getitem__(cls, item: Any, /) -> GenericAlias: ...
 
 class Maildir(Mailbox[MaildirMessage]):
     colon: str
     def __init__(
-        self, dirname: StrPath, factory: Callable[[IO[Any]], MaildirMessage] | None = None, create: bool = True
+        self, dirname: StrPath, factory: Callable[[_GetFileReturn], MaildirMessage] | None = None, create: bool = True
     ) -> None: ...
-    def add(self, message: _MessageData) -> str: ...
+    def add(self, message: _MessageData | MaildirMessage) -> str: ...
     def remove(self, key: str) -> None: ...
-    def __setitem__(self, key: str, message: _MessageData) -> None: ...
+    def __setitem__(self, key: str, message: _MessageData | MaildirMessage) -> None: ...
     def get_message(self, key: str) -> MaildirMessage: ...
     def get_bytes(self, key: str) -> bytes: ...
     def get_file(self, key: str) -> _ProxyFile[bytes]: ...
@@ -136,7 +161,7 @@ class Maildir(Mailbox[MaildirMessage]):
     def clean(self) -> None: ...
     def next(self) -> str | None: ...
 
-class _singlefileMailbox(Mailbox[_MessageT], metaclass=ABCMeta):
+class _singlefileMailbox(Mailbox[_MessageT_co], metaclass=ABCMeta):
     def add(self, message: _MessageData) -> str: ...
     def remove(self, key: str) -> None: ...
     def __setitem__(self, key: str, message: _MessageData) -> None: ...
@@ -148,20 +173,26 @@ class _singlefileMailbox(Mailbox[_MessageT], metaclass=ABCMeta):
     def flush(self) -> None: ...
     def close(self) -> None: ...
 
-class _mboxMMDF(_singlefileMailbox[_MessageT]):
-    def get_message(self, key: str) -> _MessageT: ...
+class _mboxMMDF(_singlefileMailbox[_MessageT_co]):
+    def get_message(self, key: str) -> _MessageT_co: ...
     def get_file(self, key: str, from_: bool = False) -> _PartialFile[bytes]: ...
     def get_bytes(self, key: str, from_: bool = False) -> bytes: ...
     def get_string(self, key: str, from_: bool = False) -> str: ...
 
 class mbox(_mboxMMDF[mboxMessage]):
-    def __init__(self, path: StrPath, factory: Callable[[IO[Any]], mboxMessage] | None = None, create: bool = True) -> None: ...
+    def __init__(
+        self, path: StrPath, factory: Callable[[_GetFileReturn], mboxMessage] | None = None, create: bool = True
+    ) -> None: ...
 
 class MMDF(_mboxMMDF[MMDFMessage]):
-    def __init__(self, path: StrPath, factory: Callable[[IO[Any]], MMDFMessage] | None = None, create: bool = True) -> None: ...
+    def __init__(
+        self, path: StrPath, factory: Callable[[_GetFileReturn], MMDFMessage] | None = None, create: bool = True
+    ) -> None: ...
 
 class MH(Mailbox[MHMessage]):
-    def __init__(self, path: StrPath, factory: Callable[[IO[Any]], MHMessage] | None = None, create: bool = True) -> None: ...
+    def __init__(
+        self, path: StrPath, factory: Callable[[_GetFileReturn], MHMessage] | None = None, create: bool = True
+    ) -> None: ...
     def add(self, message: _MessageData) -> str: ...
     def remove(self, key: str) -> None: ...
     def __setitem__(self, key: str, message: _MessageData) -> None: ...
@@ -184,13 +215,15 @@ class MH(Mailbox[MHMessage]):
     def pack(self) -> None: ...
 
 class Babyl(_singlefileMailbox[BabylMessage]):
-    def __init__(self, path: StrPath, factory: Callable[[IO[Any]], BabylMessage] | None = None, create: bool = True) -> None: ...
+    def __init__(
+        self, path: StrPath, factory: Callable[[_GetFileReturn], BabylMessage] | None = None, create: bool = True
+    ) -> None: ...
     def get_message(self, key: str) -> BabylMessage: ...
     def get_bytes(self, key: str) -> bytes: ...
-    def get_file(self, key: str) -> IO[bytes]: ...
+    def get_file(self, key: str) -> io.BytesIO: ...
     def get_labels(self) -> list[str]: ...
 
-class Message(email.message.Message):
+class Message(email.message.Message[str, str]):
     def __init__(self, message: _MessageData | None = None) -> None: ...
 
 class MaildirMessage(Message):
@@ -233,7 +266,7 @@ class BabylMessage(Message):
 class MMDFMessage(_mboxMMDFMessage): ...
 
 class _ProxyFile(Generic[AnyStr]):
-    def __init__(self, f: IO[AnyStr], pos: int | None = None) -> None: ...
+    def __init__(self, f: _GetFileReturn[AnyStr], pos: int | None = None) -> None: ...
     def read(self, size: int | None = None) -> AnyStr: ...
     def read1(self, size: int | None = None) -> AnyStr: ...
     def readline(self, size: int | None = None) -> AnyStr: ...
@@ -243,7 +276,7 @@ class _ProxyFile(Generic[AnyStr]):
     def seek(self, offset: int, whence: int = 0) -> None: ...
     def close(self) -> None: ...
     def __enter__(self) -> Self: ...
-    def __exit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None) -> None: ...
+    def __exit__(self, *exc: Unused) -> None: ...
     def readable(self) -> bool: ...
     def writable(self) -> bool: ...
     def seekable(self) -> bool: ...
@@ -253,7 +286,7 @@ class _ProxyFile(Generic[AnyStr]):
     def __class_getitem__(cls, item: Any, /) -> GenericAlias: ...
 
 class _PartialFile(_ProxyFile[AnyStr]):
-    def __init__(self, f: IO[AnyStr], start: int | None = None, stop: int | None = None) -> None: ...
+    def __init__(self, f: _GetFileReturn[AnyStr], start: int | None = None, stop: int | None = None) -> None: ...
 
 class Error(Exception): ...
 class NoSuchMailboxError(Error): ...