mirror of
https://github.com/davidhalter/typeshed.git
synced 2026-05-07 14:00:12 +08:00
Semyon Moroz
fb0940e6c1
* [stdlib] Deprecate many functions * Fix typo * Imporove message for asyncio/trsock * Apply suggestions from code review Co-authored-by: Alex Waygood <Alex.Waygood@Gmail.com> * Add Python before version * [pre-commit.ci] auto fixes from pre-commit.com hooks * Wrap comment lines * fix the rest --------- Co-authored-by: Alex Waygood <Alex.Waygood@Gmail.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
import asynchat
|
|
import asyncore
|
|
import socket
|
|
import sys
|
|
from collections import defaultdict
|
|
from typing import Any
|
|
from typing_extensions import TypeAlias, deprecated
|
|
|
|
if sys.version_info >= (3, 11):
|
|
__all__ = ["SMTPChannel", "SMTPServer", "DebuggingServer", "PureProxy"]
|
|
else:
|
|
__all__ = ["SMTPChannel", "SMTPServer", "DebuggingServer", "PureProxy", "MailmanProxy"]
|
|
|
|
_Address: TypeAlias = tuple[str, int] # (host, port)
|
|
|
|
class SMTPChannel(asynchat.async_chat):
|
|
COMMAND: int
|
|
DATA: int
|
|
|
|
command_size_limits: defaultdict[str, int]
|
|
smtp_server: SMTPServer
|
|
conn: socket.socket
|
|
addr: Any
|
|
received_lines: list[str]
|
|
smtp_state: int
|
|
seen_greeting: str
|
|
mailfrom: str
|
|
rcpttos: list[str]
|
|
received_data: str
|
|
fqdn: str
|
|
peer: str
|
|
|
|
command_size_limit: int
|
|
data_size_limit: int
|
|
|
|
enable_SMTPUTF8: bool
|
|
@property
|
|
def max_command_size_limit(self) -> int: ...
|
|
def __init__(
|
|
self,
|
|
server: SMTPServer,
|
|
conn: socket.socket,
|
|
addr: Any,
|
|
data_size_limit: int = 33554432,
|
|
map: asyncore._MapType | None = None,
|
|
enable_SMTPUTF8: bool = False,
|
|
decode_data: bool = False,
|
|
) -> None: ...
|
|
# base asynchat.async_chat.push() accepts bytes
|
|
def push(self, msg: str) -> None: ... # type: ignore[override]
|
|
def collect_incoming_data(self, data: bytes) -> None: ...
|
|
def found_terminator(self) -> None: ...
|
|
def smtp_HELO(self, arg: str) -> None: ...
|
|
def smtp_NOOP(self, arg: str) -> None: ...
|
|
def smtp_QUIT(self, arg: str) -> None: ...
|
|
def smtp_MAIL(self, arg: str) -> None: ...
|
|
def smtp_RCPT(self, arg: str) -> None: ...
|
|
def smtp_RSET(self, arg: str) -> None: ...
|
|
def smtp_DATA(self, arg: str) -> None: ...
|
|
def smtp_EHLO(self, arg: str) -> None: ...
|
|
def smtp_HELP(self, arg: str) -> None: ...
|
|
def smtp_VRFY(self, arg: str) -> None: ...
|
|
def smtp_EXPN(self, arg: str) -> None: ...
|
|
|
|
class SMTPServer(asyncore.dispatcher):
|
|
channel_class: type[SMTPChannel]
|
|
|
|
data_size_limit: int
|
|
enable_SMTPUTF8: bool
|
|
def __init__(
|
|
self,
|
|
localaddr: _Address,
|
|
remoteaddr: _Address,
|
|
data_size_limit: int = 33554432,
|
|
map: asyncore._MapType | None = None,
|
|
enable_SMTPUTF8: bool = False,
|
|
decode_data: bool = False,
|
|
) -> None: ...
|
|
def handle_accepted(self, conn: socket.socket, addr: Any) -> None: ...
|
|
def process_message(
|
|
self, peer: _Address, mailfrom: str, rcpttos: list[str], data: bytes | str, **kwargs: Any
|
|
) -> str | None: ...
|
|
|
|
class DebuggingServer(SMTPServer): ...
|
|
|
|
class PureProxy(SMTPServer):
|
|
def process_message(self, peer: _Address, mailfrom: str, rcpttos: list[str], data: bytes | str) -> str | None: ... # type: ignore[override]
|
|
|
|
if sys.version_info < (3, 11):
|
|
@deprecated("Deprecated since Python 3.9; removed in Python 3.11.")
|
|
class MailmanProxy(PureProxy):
|
|
def process_message(self, peer: _Address, mailfrom: str, rcpttos: list[str], data: bytes | str) -> str | None: ... # type: ignore[override]
|