Format: application/octet-stream
⬇ Download File (7.8 KB)
   1  import subprocess
   2  import sys
   3  import time
   4  from socket import socket as _socket
   5  from ssl import SSLContext, SSLSocket
   6  from types import TracebackType
   7  from typing import IO, Any, Callable, Dict, List, Optional, Pattern, Text, Tuple, Type, Union
   8  from typing_extensions import Literal
   9  
  10  # TODO: Commands should use their actual return types, not this type alias.
  11  #       E.g. Tuple[Literal["OK"], List[bytes]]
  12  _CommandResults = Tuple[str, List[Any]]
  13  
  14  _AnyResponseData = Union[List[None], List[Union[bytes, Tuple[bytes, bytes]]]]
  15  
  16  class IMAP4:
  17      error: Type[Exception] = ...
  18      abort: Type[Exception] = ...
  19      readonly: Type[Exception] = ...
  20      mustquote: Pattern[Text] = ...
  21      debug: int = ...
  22      state: str = ...
  23      literal: Optional[Text] = ...
  24      tagged_commands: Dict[bytes, Optional[List[bytes]]]
  25      untagged_responses: Dict[str, List[Union[bytes, Tuple[bytes, bytes]]]]
  26      continuation_response: str = ...
  27      is_readonly: bool = ...
  28      tagnum: int = ...
  29      tagpre: str = ...
  30      tagre: Pattern[Text] = ...
  31      welcome: bytes = ...
  32      capabilities: Tuple[str] = ...
  33      PROTOCOL_VERSION: str = ...
  34      if sys.version_info >= (3, 9):
  35          def __init__(self, host: str = ..., port: int = ..., timeout: Optional[float] = ...) -> None: ...
  36          def open(self, host: str = ..., port: int = ..., timeout: Optional[float] = ...) -> None: ...
  37      else:
  38          def __init__(self, host: str = ..., port: int = ...) -> None: ...
  39          def open(self, host: str = ..., port: int = ...) -> None: ...
  40      def __getattr__(self, attr: str) -> Any: ...
  41      host: str = ...
  42      port: int = ...
  43      sock: _socket = ...
  44      file: Union[IO[Text], IO[bytes]] = ...
  45      def read(self, size: int) -> bytes: ...
  46      def readline(self) -> bytes: ...
  47      def send(self, data: bytes) -> None: ...
  48      def shutdown(self) -> None: ...
  49      def socket(self) -> _socket: ...
  50      def recent(self) -> _CommandResults: ...
  51      def response(self, code: str) -> _CommandResults: ...
  52      def append(self, mailbox: str, flags: str, date_time: str, message: str) -> str: ...
  53      def authenticate(self, mechanism: str, authobject: Callable[[bytes], Optional[bytes]]) -> Tuple[str, str]: ...
  54      def capability(self) -> _CommandResults: ...
  55      def check(self) -> _CommandResults: ...
  56      def close(self) -> _CommandResults: ...
  57      def copy(self, message_set: str, new_mailbox: str) -> _CommandResults: ...
  58      def create(self, mailbox: str) -> _CommandResults: ...
  59      def delete(self, mailbox: str) -> _CommandResults: ...
  60      def deleteacl(self, mailbox: str, who: str) -> _CommandResults: ...
  61      if sys.version_info >= (3, 5):
  62          def enable(self, capability: str) -> _CommandResults: ...
  63          def __enter__(self) -> IMAP4: ...
  64          def __exit__(self, t: Optional[Type[BaseException]], v: Optional[BaseException], tb: Optional[TracebackType]) -> None: ...
  65      def expunge(self) -> _CommandResults: ...
  66      def fetch(self, message_set: str, message_parts: str) -> Tuple[str, _AnyResponseData]: ...
  67      def getacl(self, mailbox: str) -> _CommandResults: ...
  68      def getannotation(self, mailbox: str, entry: str, attribute: str) -> _CommandResults: ...
  69      def getquota(self, root: str) -> _CommandResults: ...
  70      def getquotaroot(self, mailbox: str) -> _CommandResults: ...
  71      def list(self, directory: str = ..., pattern: str = ...) -> Tuple[str, _AnyResponseData]: ...
  72      def login(self, user: str, password: str) -> Tuple[Literal["OK"], List[bytes]]: ...
  73      def login_cram_md5(self, user: str, password: str) -> _CommandResults: ...
  74      def logout(self) -> Tuple[str, _AnyResponseData]: ...
  75      def lsub(self, directory: str = ..., pattern: str = ...) -> _CommandResults: ...
  76      def myrights(self, mailbox: str) -> _CommandResults: ...
  77      def namespace(self) -> _CommandResults: ...
  78      def noop(self) -> Tuple[str, List[bytes]]: ...
  79      def partial(self, message_num: str, message_part: str, start: str, length: str) -> _CommandResults: ...
  80      def proxyauth(self, user: str) -> _CommandResults: ...
  81      def rename(self, oldmailbox: str, newmailbox: str) -> _CommandResults: ...
  82      def search(self, charset: Optional[str], *criteria: str) -> _CommandResults: ...
  83      def select(self, mailbox: str = ..., readonly: bool = ...) -> Tuple[str, List[Optional[bytes]]]: ...
  84      def setacl(self, mailbox: str, who: str, what: str) -> _CommandResults: ...
  85      def setannotation(self, *args: str) -> _CommandResults: ...
  86      def setquota(self, root: str, limits: str) -> _CommandResults: ...
  87      def sort(self, sort_criteria: str, charset: str, *search_criteria: str) -> _CommandResults: ...
  88      if sys.version_info >= (3,):
  89          def starttls(self, ssl_context: Optional[Any] = ...) -> Tuple[Literal["OK"], List[None]]: ...
  90      def status(self, mailbox: str, names: str) -> _CommandResults: ...
  91      def store(self, message_set: str, command: str, flags: str) -> _CommandResults: ...
  92      def subscribe(self, mailbox: str) -> _CommandResults: ...
  93      def thread(self, threading_algorithm: str, charset: str, *search_criteria: str) -> _CommandResults: ...
  94      def uid(self, command: str, *args: str) -> _CommandResults: ...
  95      def unsubscribe(self, mailbox: str) -> _CommandResults: ...
  96      if sys.version_info >= (3, 9):
  97          def unselect(self) -> _CommandResults: ...
  98      def xatom(self, name: str, *args: str) -> _CommandResults: ...
  99      def print_log(self) -> None: ...
 100  
 101  class IMAP4_SSL(IMAP4):
 102      keyfile: str = ...
 103      certfile: str = ...
 104      if sys.version_info >= (3, 9):
 105          def __init__(
 106              self,
 107              host: str = ...,
 108              port: int = ...,
 109              keyfile: Optional[str] = ...,
 110              certfile: Optional[str] = ...,
 111              ssl_context: Optional[SSLContext] = ...,
 112              timeout: Optional[float] = ...,
 113          ) -> None: ...
 114      elif sys.version_info >= (3, 3):
 115          def __init__(
 116              self,
 117              host: str = ...,
 118              port: int = ...,
 119              keyfile: Optional[str] = ...,
 120              certfile: Optional[str] = ...,
 121              ssl_context: Optional[SSLContext] = ...,
 122          ) -> None: ...
 123      else:
 124          def __init__(
 125              self, host: str = ..., port: int = ..., keyfile: Optional[str] = ..., certfile: Optional[str] = ...
 126          ) -> None: ...
 127      host: str = ...
 128      port: int = ...
 129      sock: _socket = ...
 130      sslobj: SSLSocket = ...
 131      file: IO[Any] = ...
 132      if sys.version_info >= (3, 9):
 133          def open(self, host: str = ..., port: Optional[int] = ..., timeout: Optional[float] = ...) -> None: ...
 134      else:
 135          def open(self, host: str = ..., port: Optional[int] = ...) -> None: ...
 136      def read(self, size: int) -> bytes: ...
 137      def readline(self) -> bytes: ...
 138      def send(self, data: bytes) -> None: ...
 139      def shutdown(self) -> None: ...
 140      def socket(self) -> _socket: ...
 141      def ssl(self) -> SSLSocket: ...
 142  
 143  class IMAP4_stream(IMAP4):
 144      command: str = ...
 145      def __init__(self, command: str) -> None: ...
 146      host: str = ...
 147      port: int = ...
 148      sock: _socket = ...
 149      file: IO[Any] = ...
 150      process: subprocess.Popen[bytes] = ...
 151      writefile: IO[Any] = ...
 152      readfile: IO[Any] = ...
 153      if sys.version_info >= (3, 9):
 154          def open(self, host: Optional[str] = ..., port: Optional[int] = ..., timeout: Optional[float] = ...) -> None: ...
 155      else:
 156          def open(self, host: Optional[str] = ..., port: Optional[int] = ...) -> None: ...
 157      def read(self, size: int) -> bytes: ...
 158      def readline(self) -> bytes: ...
 159      def send(self, data: bytes) -> None: ...
 160      def shutdown(self) -> None: ...
 161  
 162  class _Authenticator:
 163      mech: Callable[[bytes], bytes] = ...
 164      def __init__(self, mechinst: Callable[[bytes], bytes]) -> None: ...
 165      def process(self, data: str) -> str: ...
 166      def encode(self, inp: bytes) -> str: ...
 167      def decode(self, inp: str) -> bytes: ...
 168  
 169  def Internaldate2tuple(resp: str) -> time.struct_time: ...
 170  def Int2AP(num: int) -> str: ...
 171  def ParseFlags(resp: str) -> Tuple[str]: ...
 172  def Time2Internaldate(date_time: Union[float, time.struct_time, str]) -> str: ...
 173