Move HTTP codec

Since it makes more sense this way, because the former sounds
like a method when it's not.

optional-requirements.txt

@@ -1,4 +1,4 @@
 cryptg
-pysocks
+pproxy
 hachoir3
 pillow

readthedocs/basic/signing-in.rst

@@ -107,7 +107,7 @@ Signing In behind a Proxy
 =========================
 
 If you need to use a proxy to access Telegram,
-you will need to  `install PySocks`__ and then change:
+you will need to  `install aiosocks`__ and then change:
 
 .. code-block:: python
 
@@ -117,15 +117,29 @@ with
 
 .. code-block:: python
 
-    TelegramClient('anon', api_id, api_hash, proxy=(socks.SOCKS5, '127.0.0.1', 4444))
+    TelegramClient('anon', api_id, api_hash, proxy={
+        'host': '127.0.0.1',
+        'port': 4444
+    })
+
 
 (of course, replacing the IP and port with the IP and port of the proxy).
 
-The ``proxy=`` argument should be a tuple, a list or a dict,
-consisting of parameters described `in PySocks usage`__.
+The ``proxy=`` argument should be a dictionary
+where the following keys are allowed:
 
-.. __: https://github.com/Anorov/PySocks#installation
-.. __: https://github.com/Anorov/PySocks#usage-1
+.. code-block:: python
+
+    proxy = {
+        'host': 'localhost',    # (mandatory) proxy IP address
+        'port': 42252,          # (mandatory) proxy port number
+        'protocol': 'socks5',   # (optional) protocol to use, default socks5, allowed values: socks5, socks4
+        'username': 'foo',      # (optional) username if the proxy requires auth
+        'password': 'bar',      # (optional) password if the proxy requires auth
+        'remote_resolve': True  # (optional) whether to use remote or local resolve, default remote
+    }
+
+.. __: https://github.com/nibrag/aiosocks
 
 
 Using MTProto Proxies

readthedocs/quick-references/client-reference.rst

@@ -45,7 +45,7 @@ Base
 
     connect
     disconnect
-    is_connected
+    connected
     disconnected
     loop
 

telethon/client/account.py

@@ -54,7 +54,7 @@ class _TakeoutClient:
                 self.__success))
             if not result:
                 raise ValueError("Failed to finish the takeout.")
-            self.session.takeout_id = None
+            self._session.takeout_id = None
 
     __enter__ = helpers._sync_enter
     __exit__ = helpers._sync_exit
@@ -211,7 +211,7 @@ class AccountMethods(UserMethods):
         )
         arg_specified = (arg is not None for arg in request_kwargs.values())
 
-        if self.session.takeout_id is None or any(arg_specified):
+        if self._session.takeout_id is None or any(arg_specified):
             request = functions.account.InitTakeoutSessionRequest(
                 **request_kwargs)
         else:

telethon/client/auth.py

@@ -137,7 +137,7 @@ class AuthMethods(MessageParseMethods, UserMethods):
     async def _start(
             self, phone, password, bot_token, force_sms,
             code_callback, first_name, last_name, max_attempts):
-        if not self.is_connected():
+        if not self.connected:
             await self.connect()
 
         if await self.is_user_authorized():
@@ -258,29 +258,22 @@ class AuthMethods(MessageParseMethods, UserMethods):
 
     async def sign_in(
             self: 'TelegramClient',
-            phone: str = None,
             code: typing.Union[str, int] = None,
             *,
             password: str = None,
             bot_token: str = None,
+            phone: str = None,
             phone_code_hash: str = None) -> 'types.User':
         """
         Logs in to Telegram to an existing user or bot account.
 
         You should only use this if you are not authorized yet.
 
-        This method will send the code if it's not provided.
-
         .. note::
 
             In most cases, you should simply use `start()` and not this method.
 
         Arguments
-            phone (`str` | `int`):
-                The phone to send the code to if no code was provided,
-                or to override the phone that was previously used with
-                these requests.
-
             code (`str` | `int`):
                 The code that Telegram sent. Note that if you have sent this
                 code through the application itself it will immediately
@@ -296,30 +289,33 @@ class AuthMethods(MessageParseMethods, UserMethods):
                 This should be the hash the `@BotFather <https://t.me/BotFather>`_
                 gave you.
 
+            phone (`str` | `int`):
+                By default, the library remembers the phone passed to
+                `send_code_request`. If you are passing the code for
+                a different phone, you should set this parameter.
+
             phone_code_hash (`str`, optional):
-                The hash returned by `send_code_request`. This can be left as
-                ``None`` to use the last hash known for the phone to be used.
+                By default, the library remembers the hash that
+                `send_code_request` returned. If you are passing the
+                code for a different phone, you should set this parameter.
 
         Returns
-            The signed in user, or the information about
-            :meth:`send_code_request`.
+            The signed in user.
 
         Example
             .. code-block:: python
 
                 phone = '+34 123 123 123'
-                client.sign_in(phone)  # send code
+                client.send_code_request(phone)
 
                 code = input('enter code: ')
-                client.sign_in(phone, code)
+                client.sign_in(code)
         """
         me = await self.get_me()
         if me:
             return me
 
-        if phone and not code and not password:
-            return await self.send_code_request(phone)
-        elif code:
+        if code:
             phone, phone_code_hash = \
                 self._parse_phone_and_hash(phone, phone_code_hash)
 
@@ -335,13 +331,10 @@ class AuthMethods(MessageParseMethods, UserMethods):
         elif bot_token:
             result = await self(functions.auth.ImportBotAuthorizationRequest(
                 flags=0, bot_auth_token=bot_token,
-                api_id=self.api_id, api_hash=self.api_hash
+                api_id=self._api_id, api_hash=self._api_hash
             ))
         else:
-            raise ValueError(
-                'You must provide a phone and a code the first time, '
-                'and a password only if an RPCError was raised before.'
-            )
+            raise ValueError('You must provide a code, password or bot token')
 
         return self._on_login(result.user)
 
@@ -473,7 +466,7 @@ class AuthMethods(MessageParseMethods, UserMethods):
         if not phone_hash:
             try:
                 result = await self(functions.auth.SendCodeRequest(
-                    phone, self.api_id, self.api_hash, types.CodeSettings()))
+                    phone, self._api_id, self._api_hash, types.CodeSettings()))
             except errors.AuthRestartError:
                 return await self.send_code_request(phone, force_sms=force_sms)
 
@@ -516,7 +509,7 @@ class AuthMethods(MessageParseMethods, UserMethods):
         self._state_cache.reset()
 
         await self.disconnect()
-        self.session.delete()
+        self._session.delete()
         return True
 
     async def edit_2fa(

telethon/client/buttons.py

@@ -53,7 +53,7 @@ class ButtonMethods(UpdateMethods):
         if not utils.is_list_like(buttons):
             buttons = [[buttons]]
         elif not utils.is_list_like(buttons[0]):
-            buttons = [buttons]
+            buttons = [[b] for b in buttons]
 
         is_inline = False
         is_normal = False

telethon/client/messages.py

@@ -1,5 +1,6 @@
 import itertools
 import typing
+import warnings
 
 from .buttons import ButtonMethods
 from .messageparse import MessageParseMethods
@@ -636,6 +637,14 @@ class MessageMethods(UploadMethods, ButtonMethods, MessageParseMethods):
                 client.send_message(chat, 'A single button, with "clk1" as data',
                                     buttons=Button.inline('Click me', b'clk1'))
 
+                # Row of inline buttons (just a list)
+                client.send_message(chat, 'Look at this row',
+                                    [Button.inline('Row 1'), Button.inline('Row 2')])
+
+                # Columns of inline buttons (a list of lists)
+                client.send_message(chat, 'Look at this row',
+                                    [[Button.inline('Col 1'), Button.inline('Col 2')]])
+
                 # Matrix of inline buttons
                 client.send_message(chat, 'Pick one from this grid', buttons=[
                     [Button.inline('Left'), Button.inline('Right')],
@@ -1061,65 +1070,71 @@ class MessageMethods(UploadMethods, ButtonMethods, MessageParseMethods):
             max_id: int = None,
             clear_mentions: bool = False) -> bool:
         """
-        Marks messages as read and optionally clears mentions.
+        Deprecated, use `mark_read` instead.
+        """
+        warnings.warn('client.send_read_acknowledge is deprecated, use client.mark_read instead')
+        if max_id:
+            message = max_id
 
-        This effectively marks a message as read (or more than one) in the
-        given conversation.
+        return await self.mark_read(entity, message, clear_mentions=clear_mentions)
 
-        If neither message nor maximum ID are provided, all messages will be
-        marked as read by assuming that ``max_id = 0``.
+    async def mark_read(
+            self: 'TelegramClient',
+            entity: 'hints.EntityLike',
+            message: 'typing.Union[hints.MessageIDLike, typing.Sequence[hints.MessageIDLike]]' = (),
+            *,
+            clear_mentions: bool = False) -> bool:
+        """
+        Marks a chat as read, and optionally clears mentions.
+
+        By default, all messages will be marked as read, and mentions won't
+        be cleared. You can also specify up to which message the client has
+        read, and optionally clear mentions.
 
         Arguments
             entity (`entity`):
                 The chat where these messages are located.
 
-            message (`list` | `Message <telethon.tl.custom.message.Message>`):
-                Either a list of messages or a single message.
-
-            max_id (`int`):
-                Overrides messages, until which message should the
-                acknowledge should be sent.
+            message (`int` | `list` | `Message <telethon.tl.custom.message.Message>`):
+                Either a list of messages, a single message or an ID.
+                The chat will be marked as read up to the highest ID.
 
             clear_mentions (`bool`):
                 Whether the mention badge should be cleared (so that
                 there are no more mentions) or not for the given entity.
 
                 If no message is provided, this will be the only action
-                taken.
+                taken. If you want to mark as read *and* clear mentions,
+                pass ``0`` as the message and set this to ``True``.
 
         Example
             .. code-block:: python
 
-                client.send_read_acknowledge(last_message)
+                client.mark_read(chat)
                 # or
-                client.send_read_acknowledge(last_message_id)
+                client.mark_read(chat, some_message)
                 # or
-                client.send_read_acknowledge(messages)
+                client.mark_read(chat, clear_mentions=True)
         """
-        if max_id is None:
-            if not message:
-                max_id = 0
-            else:
-                if utils.is_list_like(message):
-                    max_id = max(msg.id for msg in message)
-                else:
-                    max_id = message.id
-
         entity = await self.get_input_entity(entity)
         if clear_mentions:
             await self(functions.messages.ReadMentionsRequest(entity))
-            if max_id is None:
+            if message == ():
                 return True
 
-        if max_id is not None:
-            if isinstance(entity, types.InputPeerChannel):
-                return await self(functions.channels.ReadHistoryRequest(
-                    utils.get_input_channel(entity), max_id=max_id))
-            else:
-                return await self(functions.messages.ReadHistoryRequest(
-                    entity, max_id=max_id))
+        if not message:
+            message = 0
+        elif utils.is_list_like(message):
+            message = max(map(utils.get_message_id, message))
+        else:
+            message = utils.get_message_id(message)
 
-        return False
+        if isinstance(entity, types.InputPeerChannel):
+            return await self(functions.channels.ReadHistoryRequest(
+                utils.get_input_channel(entity), max_id=message))
+        else:
+            return await self(functions.messages.ReadHistoryRequest(
+                entity, max_id=message))
 
     async def pin_message(
             self: 'TelegramClient',

telethon/client/telegrambaseclient.py

@@ -10,7 +10,7 @@ from .. import version, helpers, __name__ as __base_name__
 from ..crypto import rsa
 from ..entitycache import EntityCache
 from ..extensions import markdown
-from ..network import MTProtoSender, Connection, ConnectionTcpFull, TcpMTProxy
+from ..network import MTProtoSender, AsyncioConnection, BaseCodec, FullCodec
 from ..sessions import Session, SQLiteSession, MemorySession
 from ..statecache import StateCache
 from ..tl import TLObject, functions, types
@@ -67,12 +67,10 @@ class TelegramBaseClient(abc.ABC):
             By default this is ``False`` as IPv6 support is not
             too widespread yet.
 
-        proxy (`tuple` | `list` | `dict`, optional):
-            An iterable consisting of the proxy info. If `connection` is
-            one of `MTProxy`, then it should contain MTProxy credentials:
-            ``('hostname', port, 'secret')``. Otherwise, it's meant to store
-            function parameters for PySocks, like ``(type, 'hostname', port)``.
-            See https://github.com/Anorov/PySocks#usage-1 for more.
+        proxy (`dict`, optional):
+            A dictionary with information about the proxy to connect to.
+
+            See :ref:`signing-in` for details.
 
         timeout (`int` | `float`, optional):
             The timeout in seconds to be used when connecting.
@@ -169,9 +167,9 @@ class TelegramBaseClient(abc.ABC):
             api_id: int,
             api_hash: str,
             *,
-            connection: 'typing.Type[Connection]' = ConnectionTcpFull,
+            connection: 'typing.Type[BaseCodec]' = FullCodec,  # TODO rename
             use_ipv6: bool = False,
-            proxy: typing.Union[tuple, dict] = None,
+            proxy: typing.Union[str, dict] = None,
             timeout: int = 10,
             request_retries: int = 5,
             connection_retries: int =5,
@@ -246,10 +244,10 @@ class TelegramBaseClient(abc.ABC):
         # them to disk, and to save additional useful information.
         # TODO Session should probably return all cached
         #      info of entities, not just the input versions
-        self.session = session
+        self._session = session
         self._entity_cache = EntityCache()
-        self.api_id = int(api_id)
-        self.api_hash = api_hash
+        self._api_id = int(api_id)
+        self._api_hash = api_hash
 
         self._request_retries = request_retries
         self._connection_retries = connection_retries
@@ -259,16 +257,17 @@ class TelegramBaseClient(abc.ABC):
         self._auto_reconnect = auto_reconnect
 
         assert isinstance(connection, type)
-        self._connection = connection
-        init_proxy = None if not issubclass(connection, TcpMTProxy) else \
-            types.InputClientProxy(*connection.address_info(proxy))
+        self._codec = connection
+
+        # TODO set types.InputClientProxy if appropriated
+        init_proxy = None
 
         # Used on connection. Capture the variables in a lambda since
         # exporting clients need to create this InvokeWithLayerRequest.
         system = platform.uname()
         self._init_with = lambda x: functions.InvokeWithLayerRequest(
             LAYER, functions.InitConnectionRequest(
-                api_id=self.api_id,
+                api_id=self._api_id,
                 device_model=device_model or system.system or 'Unknown',
                 system_version=system_version or system.release or '1.0',
                 app_version=app_version or self.__version__,
@@ -281,7 +280,7 @@ class TelegramBaseClient(abc.ABC):
         )
 
         self._sender = MTProtoSender(
-            self.session.auth_key, self._loop,
+            self._session.auth_key, self._loop,
             loggers=self._log,
             retries=self._connection_retries,
             delay=self._retry_delay,
@@ -319,7 +318,7 @@ class TelegramBaseClient(abc.ABC):
         # Update state (for catching up after a disconnection)
         # TODO Get state from channels too
         self._state_cache = StateCache(
-            self.session.get_update_state(0), self._log)
+            self._session.get_update_state(0), self._log)
 
         # Some further state for subclasses
         self._event_builders = []
@@ -363,6 +362,18 @@ class TelegramBaseClient(abc.ABC):
         """
         return self._loop
 
+    @property
+    def session(self) -> Session:
+        """
+        The ``Session`` instance used by the client.
+
+        Example
+            .. code-block:: python
+
+                client.session.set_dc(dc_id, ip, port)
+        """
+        return self._session
+
     @property
     def disconnected(self: 'TelegramClient') -> asyncio.Future:
         """
@@ -405,36 +416,36 @@ class TelegramBaseClient(abc.ABC):
                 except OSError:
                     print('Failed to connect')
         """
-        await self._sender.connect(self._connection(
-            self.session.server_address,
-            self.session.port,
-            self.session.dc_id,
+        await self._sender.connect(AsyncioConnection(
+            self._session.server_address,
+            self._session.port,
+            self._session.dc_id,
+            codec=self._codec(),
             loop=self._loop,
             loggers=self._log,
             proxy=self._proxy
         ))
-        self.session.auth_key = self._sender.auth_key
-        self.session.save()
+        self._session.auth_key = self._sender.auth_key
+        self._session.save()
 
         await self._sender.send(self._init_with(
             functions.help.GetConfigRequest()))
 
         self._updates_handle = self._loop.create_task(self._update_loop())
 
-    def is_connected(self: 'TelegramClient') -> bool:
+    @property
+    def connected(self: 'TelegramClient') -> bool:
         """
-        Returns ``True`` if the user has connected.
-
-        This method is **not** asynchronous (don't use ``await`` on it).
+        Property which is ``True`` if the user has connected.
 
         Example
             .. code-block:: python
 
-                while client.is_connected():
+                while client.connected:
                     await asyncio.sleep(1)
         """
         sender = getattr(self, '_sender', None)
-        return sender and sender.is_connected()
+        return sender and sender.connected
 
     def disconnect(self: 'TelegramClient'):
         """
@@ -477,7 +488,7 @@ class TelegramBaseClient(abc.ABC):
 
         pts, date = self._state_cache[None]
         if pts and date:
-            self.session.set_update_state(0, types.updates.State(
+            self._session.set_update_state(0, types.updates.State(
                 pts=pts,
                 qts=0,
                 date=date,
@@ -485,7 +496,7 @@ class TelegramBaseClient(abc.ABC):
                 unread_count=0
             ))
 
-        self.session.close()
+        self._session.close()
 
     async def _disconnect(self: 'TelegramClient'):
         """
@@ -505,12 +516,12 @@ class TelegramBaseClient(abc.ABC):
         self._log[__name__].info('Reconnecting to new data center %s', new_dc)
         dc = await self._get_dc(new_dc)
 
-        self.session.set_dc(dc.id, dc.ip_address, dc.port)
+        self._session.set_dc(dc.id, dc.ip_address, dc.port)
         # auth_key's are associated with a server, which has now changed
         # so it's not valid anymore. Set to None to force recreating it.
         self._sender.auth_key.key = None
-        self.session.auth_key = None
-        self.session.save()
+        self._session.auth_key = None
+        self._session.save()
         await self._disconnect()
         return await self.connect()
 
@@ -519,8 +530,8 @@ class TelegramBaseClient(abc.ABC):
         Callback from the sender whenever it needed to generate a
         new authorization key. This means we are not authorized.
         """
-        self.session.auth_key = auth_key
-        self.session.save()
+        self._session.auth_key = auth_key
+        self._session.save()
 
     # endregion
 
@@ -623,13 +634,13 @@ class TelegramBaseClient(abc.ABC):
         session = self._exported_sessions.get(cdn_redirect.dc_id)
         if not session:
             dc = await self._get_dc(cdn_redirect.dc_id, cdn=True)
-            session = self.session.clone()
+            session = self._session.clone()
             await session.set_dc(dc.id, dc.ip_address, dc.port)
             self._exported_sessions[cdn_redirect.dc_id] = session
 
         self._log[__name__].info('Creating new CDN client')
         client = TelegramBareClient(
-            session, self.api_id, self.api_hash,
+            session, self._api_id, self._api_hash,
             proxy=self._sender.connection.conn.proxy,
             timeout=self._sender.connection.get_timeout()
         )

telethon/client/updates.py

@@ -225,7 +225,7 @@ class UpdateMethods(UserMethods):
         if not pts:
             return
 
-        self.session.catching_up = True
+        self._session.catching_up = True
         try:
             while True:
                 d = await self(functions.updates.GetDifferenceRequest(
@@ -275,7 +275,7 @@ class UpdateMethods(UserMethods):
         finally:
             # TODO Save new pts to session
             self._state_cache._pts_date = (pts, date)
-            self.session.catching_up = False
+            self._session.catching_up = False
 
     # endregion
 
@@ -285,7 +285,7 @@ class UpdateMethods(UserMethods):
     # the order that the updates arrive in to update the pts and date to
     # be always-increasing. There is also no need to make this async.
     def _handle_update(self: 'TelegramClient', update):
-        self.session.process_entities(update)
+        self._session.process_entities(update)
         self._entity_cache.add(update)
 
         if isinstance(update, (types.Updates, types.UpdatesCombined)):
@@ -323,7 +323,7 @@ class UpdateMethods(UserMethods):
     async def _update_loop(self: 'TelegramClient'):
         # Pings' ID don't really need to be secure, just "random"
         rnd = lambda: random.randrange(-2**63, 2**63)
-        while self.is_connected():
+        while self.connected:
             try:
                 await asyncio.wait_for(
                     self.disconnected, timeout=60, loop=self._loop
@@ -347,7 +347,7 @@ class UpdateMethods(UserMethods):
             # inserted because this is a rather expensive operation
             # (default's sqlite3 takes ~0.1s to commit changes). Do
             # it every minute instead. No-op if there's nothing new.
-            self.session.save()
+            self._session.save()
 
             # We need to send some content-related request at least hourly
             # for Telegram to keep delivering updates, otherwise they will
@@ -425,7 +425,7 @@ class UpdateMethods(UserMethods):
                 )
                 break
             except Exception as e:
-                if not isinstance(e, asyncio.CancelledError) or self.is_connected():
+                if not isinstance(e, asyncio.CancelledError) or self.connected:
                     name = getattr(callback, '__name__', repr(callback))
                     self._log[__name__].exception('Unhandled exception on %s',
                                                   name)

telethon/client/uploads.py

@@ -24,18 +24,6 @@ if typing.TYPE_CHECKING:
     from .telegramclient import TelegramClient
 
 
-class _CacheType:
-    """Like functools.partial but pretends to be the wrapped class."""
-    def __init__(self, cls):
-        self._cls = cls
-
-    def __call__(self, *args, **kwargs):
-        return self._cls(*args, file_reference=b'', **kwargs)
-
-    def __eq__(self, other):
-        return self._cls == other
-
-
 def _resize_photo_if_needed(
         file, is_image, width=1280, height=1280, background=(255, 255, 255)):
 
@@ -99,7 +87,6 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
             reply_to: 'hints.MessageIDLike' = None,
             attributes: 'typing.Sequence[types.TypeDocumentAttribute]' = None,
             thumb: 'hints.FileLike' = None,
-            allow_cache: bool = True,
             parse_mode: str = (),
             voice_note: bool = False,
             video_note: bool = False,
@@ -156,8 +143,10 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
                 To send an album, you should provide a list in this parameter.
 
                 If a list or similar is provided, the files in it will be
-                sent as an album in the order in which they appear, sliced
-                in chunks of 10 if more than 10 are given.
+                sent as an album in the order in which they appear. Currently,
+                only up to 10 files are allowed, and you're responsible for
+                making sure that they are all allowed inside albums (e.g.
+                only photos or only videos, no other documents in between).
 
             caption (`str`, optional):
                 Optional caption for the sent media message. When sending an
@@ -188,12 +177,6 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
                 Successful thumbnails were files below 20kb and 200x200px.
                 Width/height and dimensions/size ratios may be important.
 
-            allow_cache (`bool`, optional):
-                Whether to allow using the cached version stored in the
-                database or not. Defaults to ``True`` to avoid re-uploads.
-                Must be ``False`` if you wish to use different attributes
-                or thumb than those that were used when the file was cached.
-
             parse_mode (`object`, optional):
                 See the `TelegramClient.parse_mode
                 <telethon.client.messageparse.MessageParseMethods.parse_mode>`
@@ -203,16 +186,10 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
             voice_note (`bool`, optional):
                 If ``True`` the audio will be sent as a voice note.
 
-                Set `allow_cache` to ``False`` if you sent the same file
-                without this setting before for it to work.
-
             video_note (`bool`, optional):
                 If ``True`` the video will be sent as a video note,
                 also known as a round video message.
 
-                Set `allow_cache` to ``False`` if you sent the same file
-                without this setting before for it to work.
-
             buttons (`list`, `custom.Button <telethon.tl.custom.button.Button>`, :tl:`KeyboardButton`):
                 The matrix (list of lists), row list or button to be shown
                 after sending the message. This parameter will only work if
@@ -267,52 +244,14 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
         if not caption:
             caption = ''
 
-        # First check if the user passed an iterable, in which case
-        # we may want to send as an album if all are photo files.
+        # First check if the user passed an iterable -> send as album
         if utils.is_list_like(file):
-            image_captions = []
-            document_captions = []
-            if utils.is_list_like(caption):
-                captions = caption
-            else:
-                captions = [caption]
-
             # TODO Fix progress_callback
-            images = []
-            if force_document:
-                documents = file
-            else:
-                documents = []
-                for doc, cap in itertools.zip_longest(file, captions):
-                    if utils.is_image(doc):
-                        images.append(doc)
-                        image_captions.append(cap)
-                    else:
-                        documents.append(doc)
-                        document_captions.append(cap)
-
-            result = []
-            while images:
-                result += await self._send_album(
-                    entity, images[:10], caption=image_captions[:10],
-                    progress_callback=progress_callback, reply_to=reply_to,
-                    parse_mode=parse_mode, silent=silent
-                )
-                images = images[10:]
-                image_captions = image_captions[10:]
-
-            for doc, cap in zip(documents, captions):
-                result.append(await self.send_file(
-                    entity, doc, allow_cache=allow_cache,
-                    caption=cap, force_document=force_document,
-                    progress_callback=progress_callback, reply_to=reply_to,
-                    attributes=attributes, thumb=thumb, voice_note=voice_note,
-                    video_note=video_note, buttons=buttons, silent=silent,
-                    supports_streaming=supports_streaming,
-                    **kwargs
-                ))
-
-            return result
+            return await self._send_album(
+                entity, file, caption=file,
+                progress_callback=progress_callback, reply_to=reply_to,
+                parse_mode=parse_mode, silent=silent
+            )
 
         entity = await self.get_input_entity(entity)
         reply_to = utils.get_message_id(reply_to)
@@ -328,7 +267,7 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
         file_handle, media, image = await self._file_to_media(
             file, force_document=force_document,
             progress_callback=progress_callback,
-            attributes=attributes,  allow_cache=allow_cache, thumb=thumb,
+            attributes=attributes, thumb=thumb,
             voice_note=voice_note, video_note=video_note,
             supports_streaming=supports_streaming
         )
@@ -343,7 +282,6 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
             entities=msg_entities, reply_markup=markup, silent=silent
         )
         msg = self._get_response_message(request, await self(request), entity)
-        await self._cache_media(msg, file, file_handle, image=image)
 
         return msg
 
@@ -351,15 +289,6 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
                           progress_callback=None, reply_to=None,
                           parse_mode=(), silent=None):
         """Specialized version of .send_file for albums"""
-        # We don't care if the user wants to avoid cache, we will use it
-        # anyway. Why? The cached version will be exactly the same thing
-        # we need to produce right now to send albums (uploadMedia), and
-        # cache only makes a difference for documents where the user may
-        # want the attributes used on them to change.
-        #
-        # In theory documents can be sent inside the albums but they appear
-        # as different messages (not inside the album), and the logic to set
-        # the attributes/avoid cache is already written in .send_file().
         entity = await self.get_input_entity(entity)
         if not utils.is_list_like(caption):
             caption = (caption,)
@@ -370,7 +299,6 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
 
         reply_to = utils.get_message_id(reply_to)
 
-        # Need to upload the media first, but only if they're not cached yet
         media = []
         for file in files:
             # Albums want :tl:`InputMedia` which, in theory, includes
@@ -382,10 +310,12 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
                 r = await self(functions.messages.UploadMediaRequest(
                     entity, media=fm
                 ))
-                self.session.cache_file(
-                    fh.md5, fh.size, utils.get_input_photo(r.photo))
-
                 fm = utils.get_input_media(r.photo)
+            elif isinstance(fm, types.InputMediaUploadedDocument):
+                r = await self(functions.messages.UploadMediaRequest(
+                    entity, media=fm
+                ))
+                fm = utils.get_input_media(r.document)
 
             if captions:
                 caption, msg_entities = captions.pop()
@@ -414,13 +344,14 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
         # Sent photo IDs -> messages
         return [messages[m.media.id.id] for m in media]
 
+    # TODO Offer a way to easily save media for later use, to replace old caching system
+
     async def upload_file(
             self: 'TelegramClient',
             file: 'hints.FileLike',
             *,
             part_size_kb: float = None,
             file_name: str = None,
-            use_cache: type = None,
             progress_callback: 'hints.ProgressCallback' = None) -> 'types.TypeInputFile':
         """
         Uploads a file to Telegram's servers, without sending it.
@@ -449,13 +380,6 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
                 If not specified, the name will be taken from the ``file``
                 and if this is not a ``str``, it will be ``"unnamed"``.
 
-            use_cache (`type`, optional):
-                The type of cache to use (currently either :tl:`InputDocument`
-                or :tl:`InputPhoto`). If present and the file is small enough
-                to need the MD5, it will be checked against the database,
-                and if a match is found, the upload won't be made. Instead,
-                an instance of type ``use_cache`` will be returned.
-
             progress_callback (`callable`, optional):
                 A callback function accepting two parameters:
                 ``(sent bytes, total)``.
@@ -537,19 +461,11 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
         hash_md5 = hashlib.md5()
         if not is_large:
             # Calculate the MD5 hash before anything else.
-            # As this needs to be done always for small files,
-            # might as well do it before anything else and
-            # check the cache.
+            # This needs to be done always for small files.
             if isinstance(file, str):
                 with open(file, 'rb') as stream:
                     file = stream.read()
             hash_md5.update(file)
-            if use_cache:
-                cached = self.session.get_file(
-                    hash_md5.digest(), file_size, cls=_CacheType(use_cache)
-                )
-                if cached:
-                    return cached
 
         part_count = (file_size + part_size - 1) // part_size
         self._log[__name__].info('Uploading file of %d bytes in %d chunks of %d',
@@ -592,7 +508,7 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
     async def _file_to_media(
             self, file, force_document=False,
             progress_callback=None, attributes=None, thumb=None,
-            allow_cache=True, voice_note=False, video_note=False,
+            voice_note=False, video_note=False,
             supports_streaming=False, mime_type=None, as_image=None):
         if not file:
             return None, None, None
@@ -626,12 +542,10 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
 
         media = None
         file_handle = None
-        use_cache = types.InputPhoto if as_image else types.InputDocument
         if not isinstance(file, str) or os.path.isfile(file):
             file_handle = await self.upload_file(
                 _resize_photo_if_needed(file, as_image),
-                progress_callback=progress_callback,
-                use_cache=use_cache if allow_cache else None
+                progress_callback=progress_callback
             )
         elif re.match('https?://', file):
             if as_image:
@@ -652,12 +566,6 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
                 'Failed to convert {} to media. Not an existing file, '
                 'an HTTP URL or a valid bot-API-like file ID'.format(file)
             )
-        elif isinstance(file_handle, use_cache):
-            # File was cached, so an instance of use_cache was returned
-            if as_image:
-                media = types.InputMediaPhoto(file_handle)
-            else:
-                media = types.InputMediaDocument(file_handle)
         elif as_image:
             media = types.InputMediaUploadedPhoto(file_handle)
         else:
@@ -685,16 +593,4 @@ class UploadMethods(ButtonMethods, MessageParseMethods, UserMethods):
             )
         return file_handle, media, as_image
 
-    async def _cache_media(self: 'TelegramClient', msg, file, file_handle, image):
-        if file and msg and isinstance(file_handle,
-                                       custom.InputSizedFile):
-            # There was a response message and we didn't use cached
-            # version, so cache whatever we just sent to the database.
-            md5, size = file_handle.md5, file_handle.size
-            if image:
-                to_cache = utils.get_input_photo(msg.media.photo)
-            else:
-                to_cache = utils.get_input_document(msg.media.document)
-            self.session.cache_file(md5, size, to_cache)
-
     # endregion

telethon/client/users.py

@@ -52,7 +52,7 @@ class UserMethods(TelegramBaseClient):
                             exceptions.append(e)
                             results.append(None)
                             continue
-                        self.session.process_entities(result)
+                        self._session.process_entities(result)
                         self._entity_cache.add(result)
                         exceptions.append(None)
                         results.append(result)
@@ -63,7 +63,7 @@ class UserMethods(TelegramBaseClient):
                         return results
                 else:
                     result = await future
-                    self.session.process_entities(result)
+                    self._session.process_entities(result)
                     self._entity_cache.add(result)
                     return result
             except (errors.ServerError, errors.RpcCallFailError,
@@ -377,7 +377,7 @@ class UserMethods(TelegramBaseClient):
 
         # No InputPeer, cached peer, or known string. Fetch from disk cache
         try:
-            return self.session.get_input_entity(peer)
+            return self._session.get_input_entity(peer)
         except ValueError:
             pass
 
@@ -513,7 +513,7 @@ class UserMethods(TelegramBaseClient):
             try:
                 # Nobody with this username, maybe it's an exact name/title
                 return await self.get_entity(
-                    self.session.get_input_entity(string))
+                    self._session.get_input_entity(string))
             except ValueError:
                 pass
 

telethon/network/__init__.py

@@ -5,10 +5,5 @@ with Telegram's servers and the protocol used (TCP full, abridged, etc.).
 from .mtprotoplainsender import MTProtoPlainSender
 from .authenticator import do_authentication
 from .mtprotosender import MTProtoSender
-from .connection import (
-    Connection,
-    ConnectionTcpFull, ConnectionTcpIntermediate, ConnectionTcpAbridged,
-    ConnectionTcpObfuscated, ConnectionTcpMTProxyAbridged,
-    ConnectionTcpMTProxyIntermediate,
-    ConnectionTcpMTProxyRandomizedIntermediate, ConnectionHttp, TcpMTProxy
-)
+from .codec import BaseCodec, FullCodec, IntermediateCodec, AbridgedCodec
+from .connection import BaseConnection, AsyncioConnection

telethon/network/codec/__init__.py

@@ -0,0 +1,5 @@
+from .basecodec import BaseCodec
+from .fullcodec import FullCodec
+from .intermediatecodec import IntermediateCodec
+from .abridgedcodec import AbridgedCodec
+from .httpcodec import HttpCodec

telethon/network/codec/abridgedcodec.py

@@ -0,0 +1,37 @@
+import struct
+
+from .basecodec import BaseCodec
+
+
+class AbridgedCodec(BaseCodec):
+    """
+    This is the mode with the lowest overhead, as it will
+    only require 1 byte if the packet length is less than
+    508 bytes (127 << 2, which is very common).
+    """
+    @staticmethod
+    def header_length():
+        return 1
+
+    @staticmethod
+    def tag():
+        return b'\xef'  # note: obfuscated tag is this 4 times
+
+    def encode_packet(self, data, ip, port):
+        length = len(data) >> 2
+        if length < 127:
+            length = struct.pack('B', length)
+        else:
+            length = b'\x7f' + int.to_bytes(length, 3, 'little')
+
+        return length + data
+
+    def decode_header(self, header):
+        if len(header) == 4:
+            length = struct.unpack('<i', header[1:] + b'\0')[0]
+        else:
+            length = struct.unpack('<B', header)[0]
+            if length >= 127:
+                return -3  # needs 3 more bytes
+
+        return length << 2

telethon/network/codec/basecodec.py

@@ -0,0 +1,53 @@
+import abc
+
+
+class BaseCodec(abc.ABC):
+    @staticmethod
+    @abc.abstractmethod
+    def header_length():
+        """
+        Returns the initial length of the header.
+        """
+        raise NotImplementedError
+
+    @staticmethod
+    @abc.abstractmethod
+    def tag():
+        """
+        The bytes tag that identifies the codec.
+
+        It may be ``None`` if there is no tag to send.
+
+        The tag will be sent upon successful connections to the
+        server so that it knows which codec we will be using next.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def encode_packet(self, data, ip, port):
+        """
+        Encodes the given data with the current codec instance.
+
+        Should return header + body.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def decode_header(self, header):
+        """
+        Decodes the header.
+
+        Should return the length of the body as a positive number.
+
+        If more data is needed, a ``-length`` should be returned, where
+        ``length`` is how much more data is needed for the full header.
+        """
+        raise NotImplementedError
+
+    def decode_body(self, header, body):
+        """
+        Decodes the body.
+
+        The default implementation returns ``body``.
+        """
+        return body

telethon/network/codec/fullcodec.py

@@ -1,43 +1,45 @@
 import struct
-from zlib import crc32
+import zlib
 
-from .connection import Connection, PacketCodec
+from .basecodec import BaseCodec
 from ...errors import InvalidChecksumError
 
 
-class FullPacketCodec(PacketCodec):
-    tag = None
-
-    def __init__(self, connection):
-        super().__init__(connection)
+class FullCodec(BaseCodec):
+    """
+    Default Telegram codec. Sends 12 additional bytes and
+    needs to calculate the CRC value of the packet itself.
+    """
+    def __init__(self):
         self._send_counter = 0  # Important or Telegram won't reply
 
-    def encode_packet(self, data):
+    @staticmethod
+    def header_length():
+        return 8
+
+    @staticmethod
+    def tag():
+        return None
+
+    def encode_packet(self, data, ip, port):
         # https://core.telegram.org/mtproto#tcp-transport
         # total length, sequence number, packet and checksum (CRC32)
         length = len(data) + 12
         data = struct.pack('<ii', length, self._send_counter) + data
-        crc = struct.pack('<I', crc32(data))
+        crc = struct.pack('<I', zlib.crc32(data))
         self._send_counter += 1
         return data + crc
 
-    async def read_packet(self, reader):
-        packet_len_seq = await reader.readexactly(8)  # 4 and 4
-        packet_len, seq = struct.unpack('<ii', packet_len_seq)
-        body = await reader.readexactly(packet_len - 8)
+    def decode_header(self, header):
+        length, seq = struct.unpack('<ii', header)
+        return length - 8
+
+    def decode_body(self, header, body):
         checksum = struct.unpack('<I', body[-4:])[0]
         body = body[:-4]
 
-        valid_checksum = crc32(packet_len_seq + body)
+        valid_checksum = zlib.crc32(header + body)
         if checksum != valid_checksum:
             raise InvalidChecksumError(checksum, valid_checksum)
 
         return body
-
-
-class ConnectionTcpFull(Connection):
-    """
-    Default Telegram mode. Sends 12 additional bytes and
-    needs to calculate the CRC value of the packet itself.
-    """
-    packet_codec = FullPacketCodec

telethon/network/codec/httpcodec.py

@@ -0,0 +1,33 @@
+from .basecodec import BaseCodec
+
+
+SSL_PORT = 443
+
+
+class HttpCodec(BaseCodec):
+    @staticmethod
+    def header_length():
+        return 4
+
+    @staticmethod
+    def tag():
+        return None
+
+    def encode_packet(self, data, ip, port):
+        return ('POST /api HTTP/1.1\r\n'
+                'Host: {}:{}\r\n'
+                'Content-Type: application/x-www-form-urlencoded\r\n'
+                'Connection: keep-alive\r\n'
+                'Keep-Alive: timeout=100000, max=10000000\r\n'
+                'Content-Length: {}\r\n\r\n'
+                .format(ip, port, len(data))
+                .encode('ascii') + data)
+
+    def decode_header(self, header):
+        if not header.endswith(b'\r\n\r\n'):
+            return -1
+
+        header = header.lower()
+        start = header.index(b'content-length: ') + 16
+        print(header)
+        return int(header[start:header.index(b'\r', start)])

telethon/network/codec/intermediatecodec.py

@@ -0,0 +1,47 @@
+import struct
+import random
+import os
+
+from .basecodec import BaseCodec
+
+
+class IntermediateCodec(BaseCodec):
+    """
+    Intermediate mode between `FullCodec` and `AbridgedCodec`.
+    Always sends 4 extra bytes for the packet length.
+    """
+    @staticmethod
+    def header_length():
+        return 4
+
+    @staticmethod
+    def tag():
+        return b'\xee\xee\xee\xee'  # same as obfuscate tag
+
+    def encode_packet(self, data, ip, port):
+        return struct.pack('<i', len(data)) + data
+
+    def decode_header(self, header):
+        return struct.unpack('<i', header)[0]
+
+
+class RandomizedIntermediateCodec(IntermediateCodec):
+    """
+    Data packets are aligned to 4 bytes. This codec adds random
+    bytes of size from 0 to 3 bytes, which are ignored by decoder.
+    """
+    tag = None
+    obfuscate_tag = b'\xdd\xdd\xdd\xdd'
+
+    def encode_packet(self, data, ip, port):
+        pad_size = random.randint(0, 3)
+        padding = os.urandom(pad_size)
+        return super().encode_packet(data + padding)
+
+    async def read_packet(self, reader):
+        raise NotImplementedError(':)')
+        packet_with_padding = await super().read_packet(reader)
+        pad_size = len(packet_with_padding) % 4
+        if pad_size > 0:
+            return packet_with_padding[:-pad_size]
+        return packet_with_padding

telethon/network/connection/__init__.py

@@ -1,12 +1,2 @@
-from .connection import Connection
-from .tcpfull import ConnectionTcpFull
-from .tcpintermediate import ConnectionTcpIntermediate
-from .tcpabridged import ConnectionTcpAbridged
-from .tcpobfuscated import ConnectionTcpObfuscated
-from .tcpmtproxy import (
-    TcpMTProxy,
-    ConnectionTcpMTProxyAbridged,
-    ConnectionTcpMTProxyIntermediate,
-    ConnectionTcpMTProxyRandomizedIntermediate
-)
-from .http import ConnectionHttp
+from .baseconnection import BaseConnection
+from .asyncioconnection import AsyncioConnection

telethon/network/connection/asyncioconnection.py

@@ -0,0 +1,169 @@
+import abc
+import asyncio
+import socket
+import ssl as ssl_mod
+import sys
+
+from ...errors import InvalidChecksumError
+from ... import helpers
+from .baseconnection import BaseConnection
+from ..codec import HttpCodec
+
+
+class AsyncioConnection(BaseConnection):
+    """
+    The `AsyncioConnection` class is a wrapper around ``asyncio.open_connection``.
+
+    Subclasses will implement different transport modes as atomic operations,
+    which this class eases doing since the exposed interface simply puts and
+    gets complete data payloads to and from queues.
+
+    The only error that will raise from send and receive methods is
+    ``ConnectionError``, which will raise when attempting to send if
+    the client is disconnected (includes remote disconnections).
+    """
+    # this static attribute should be redefined by `Connection` subclasses and
+    # should be one of `PacketCodec` implementations
+    packet_codec = None
+
+    def __init__(self, ip, port, dc_id, *, loop, codec, loggers, proxy=None):
+        super().__init__(ip, port, loop=loop, codec=codec)
+        self._dc_id = dc_id  # only for MTProxy, it's an abstraction leak
+        self._log = loggers[__name__]
+        self._proxy = proxy
+        self._reader = None
+        self._writer = None
+        self._connected = False
+        self._obfuscation = None  # TcpObfuscated and MTProxy
+
+    async def _connect(self, timeout=None):
+        if not self._proxy:
+            connect_coroutine = asyncio.open_connection(
+                self._ip, self._port, loop=self._loop)
+        else:
+            import aiosocks
+
+            auth = None
+            proto = self._proxy.get('protocol', 'socks5').lower()
+            if proto == 'socks5':
+                proxy = aiosocks.Socks5Addr(self._proxy['host'], self._proxy['port'])
+                if 'username' in self._proxy:
+                    auth = aiosocks.Socks5Auth(self._proxy['username'], self._proxy['password'])
+
+            elif proto == 'socks4':
+                proxy = aiosocks.Socks4Addr(self._proxy['host'], self._proxy['port'])
+                if 'username' in self._proxy:
+                    auth = aiosocks.Socks4Auth(self._proxy['username'])
+
+            else:
+                raise ValueError('Unsupported proxy protocol {}'.format(self._proxy['protocol']))
+
+            connect_coroutine = aiosocks.open_connection(
+                proxy=proxy,
+                proxy_auth=auth,
+                dst=(self._ip, self._port),
+                remote_resolve=self._proxy.get('remote_resolve', True),
+                loop=self._loop
+            )
+
+        self._reader, self._writer = await asyncio.wait_for(
+            connect_coroutine,
+            loop=self._loop, timeout=timeout
+        )
+
+        self._codec.__init__()  # reset the codec
+        if self._codec.tag():
+            await self._send(self._codec.tag())
+
+    @property
+    def connected(self):
+        return self._connected
+
+    async def connect(self, timeout=None):
+        """
+        Establishes a connection with the server.
+        """
+        await self._connect(timeout=timeout)
+        self._connected = True
+
+    async def disconnect(self):
+        """
+        Disconnects from the server, and clears
+        pending outgoing and incoming messages.
+        """
+        self._connected = False
+
+        if self._writer:
+            self._writer.close()
+            if sys.version_info >= (3, 7):
+                try:
+                    await self._writer.wait_closed()
+                except Exception as e:
+                    # Seen OSError: No route to host
+                    # Disconnecting should never raise
+                    self._log.warning('Unhandled %s on disconnect: %s', type(e), e)
+
+    async def _send(self, data):
+        self._writer.write(data)
+        await self._writer.drain()
+
+    async def _recv(self, length):
+        return await self._reader.readexactly(length)
+
+
+class Connection(abc.ABC):
+    pass
+
+
+class ObfuscatedConnection(Connection):
+    """
+    Base class for "obfuscated" connections ("obfuscated2", "mtproto proxy")
+    """
+    """
+    This attribute should be redefined by subclasses
+    """
+    obfuscated_io = None
+
+    def _init_conn(self):
+        self._obfuscation = self.obfuscated_io(self)
+        self._writer.write(self._obfuscation.header)
+
+    def _send(self, data):
+        self._obfuscation.write(self._codec.encode_packet(data))
+
+    async def _recv(self):
+        return await self._codec.read_packet(self._obfuscation)
+
+
+class PacketCodec(abc.ABC):
+    """
+    Base class for packet codecs
+    """
+
+    """
+    This attribute should be re-defined by subclass to define if some
+    "magic bytes" should be sent to server right after conection is made to
+    signal which protocol will be used
+    """
+    tag = None
+
+    def __init__(self, connection):
+        """
+        Codec is created when connection is just made.
+        """
+        self._conn = connection
+
+    @abc.abstractmethod
+    def encode_packet(self, data):
+        """
+        Encodes single packet and returns encoded bytes.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def read_packet(self, reader):
+        """
+        Reads single packet from `reader` object that should have
+        `readexactly(n)` method.
+        """
+        raise NotImplementedError

telethon/network/connection/baseconnection.py

@@ -0,0 +1,81 @@
+import abc
+import asyncio
+
+from ..codec import BaseCodec
+
+
+class BaseConnection(abc.ABC):
+    """
+    The base connection class.
+
+    It offers atomic send and receive methods.
+
+    Subclasses are only responsible of sending and receiving data,
+    since this base class already makes use of the given codec for
+    correctly adapting the data.
+    """
+    def __init__(self, ip: str, port: int, *, loop: asyncio.AbstractEventLoop, codec: BaseCodec):
+        self._ip = ip
+        self._port = port
+        self._loop = loop
+        self._codec = codec
+        self._send_lock = asyncio.Lock(loop=loop)
+        self._recv_lock = asyncio.Lock(loop=loop)
+
+    @property
+    @abc.abstractmethod
+    def connected(self):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def connect(self):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def disconnect(self):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def _send(self, data):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def _recv(self, length):
+        raise NotImplementedError
+
+    async def send(self, data):
+        if not self.connected:
+            raise ConnectionError('Not connected')
+
+        # TODO Handle asyncio.CancelledError, IOError, Exception
+        data = self._codec.encode_packet(data, self._ip, self._port)
+        async with self._send_lock:
+            return await self._send(data)
+
+    async def recv(self):
+        if not self.connected:
+            raise ConnectionError('Not connected')
+
+        # TODO Handle asyncio.CancelledError, asyncio.IncompleteReadError,
+        #      IOError, InvalidChecksumError, Exception properly
+        await self._recv_lock.acquire()
+        try:
+            header = await self._recv(self._codec.header_length())
+
+            length = self._codec.decode_header(header)
+            while length < 0:
+                header += await self._recv(-length)
+                length = self._codec.decode_header(header)
+
+            body = await self._recv(length)
+            return self._codec.decode_body(header, body)
+        except Exception:
+            raise ConnectionError
+        finally:
+            self._recv_lock.release()
+
+    def __str__(self):
+        return '{}:{}/{}'.format(
+            self._ip, self._port,
+            self.__class__.__name__.replace('Connection', '')
+        )

telethon/network/connection/connection.py

@@ -1,271 +0,0 @@
-import abc
-import asyncio
-import socket
-import ssl as ssl_mod
-import sys
-
-from ...errors import InvalidChecksumError
-from ... import helpers
-
-
-class Connection(abc.ABC):
-    """
-    The `Connection` class is a wrapper around ``asyncio.open_connection``.
-
-    Subclasses will implement different transport modes as atomic operations,
-    which this class eases doing since the exposed interface simply puts and
-    gets complete data payloads to and from queues.
-
-    The only error that will raise from send and receive methods is
-    ``ConnectionError``, which will raise when attempting to send if
-    the client is disconnected (includes remote disconnections).
-    """
-    # this static attribute should be redefined by `Connection` subclasses and
-    # should be one of `PacketCodec` implementations
-    packet_codec = None
-
-    def __init__(self, ip, port, dc_id, *, loop, loggers, proxy=None):
-        self._ip = ip
-        self._port = port
-        self._dc_id = dc_id  # only for MTProxy, it's an abstraction leak
-        self._loop = loop
-        self._log = loggers[__name__]
-        self._proxy = proxy
-        self._reader = None
-        self._writer = None
-        self._connected = False
-        self._send_task = None
-        self._recv_task = None
-        self._codec = None
-        self._obfuscation = None  # TcpObfuscated and MTProxy
-        self._send_queue = asyncio.Queue(1)
-        self._recv_queue = asyncio.Queue(1)
-
-    async def _connect(self, timeout=None, ssl=None):
-        if not self._proxy:
-            self._reader, self._writer = await asyncio.wait_for(
-                asyncio.open_connection(
-                    self._ip, self._port, loop=self._loop, ssl=ssl),
-                loop=self._loop, timeout=timeout
-            )
-        else:
-            import socks
-            if ':' in self._ip:
-                mode, address = socket.AF_INET6, (self._ip, self._port, 0, 0)
-            else:
-                mode, address = socket.AF_INET, (self._ip, self._port)
-
-            s = socks.socksocket(mode, socket.SOCK_STREAM)
-            if isinstance(self._proxy, dict):
-                s.set_proxy(**self._proxy)
-            else:
-                s.set_proxy(*self._proxy)
-
-            s.setblocking(False)
-            await asyncio.wait_for(
-                self._loop.sock_connect(s, address),
-                timeout=timeout,
-                loop=self._loop
-            )
-            if ssl:
-                s.settimeout(timeout)
-                s = ssl_mod.wrap_socket(
-                    s,
-                    do_handshake_on_connect=True,
-                    ssl_version=ssl_mod.PROTOCOL_SSLv23,
-                    ciphers='ADH-AES256-SHA'
-                )
-                s.setblocking(False)
-
-            self._reader, self._writer = \
-                await asyncio.open_connection(sock=s, loop=self._loop)
-
-        self._codec = self.packet_codec(self)
-        self._init_conn()
-        await self._writer.drain()
-
-    async def connect(self, timeout=None, ssl=None):
-        """
-        Establishes a connection with the server.
-        """
-        await self._connect(timeout=timeout, ssl=ssl)
-        self._connected = True
-
-        self._send_task = self._loop.create_task(self._send_loop())
-        self._recv_task = self._loop.create_task(self._recv_loop())
-
-    async def disconnect(self):
-        """
-        Disconnects from the server, and clears
-        pending outgoing and incoming messages.
-        """
-        self._connected = False
-
-        await helpers._cancel(
-            self._log,
-            send_task=self._send_task,
-            recv_task=self._recv_task
-        )
-
-        if self._writer:
-            self._writer.close()
-            if sys.version_info >= (3, 7):
-                try:
-                    await self._writer.wait_closed()
-                except Exception as e:
-                    # Seen OSError: No route to host
-                    # Disconnecting should never raise
-                    self._log.warning('Unhandled %s on disconnect: %s', type(e), e)
-
-    def send(self, data):
-        """
-        Sends a packet of data through this connection mode.
-
-        This method returns a coroutine.
-        """
-        if not self._connected:
-            raise ConnectionError('Not connected')
-
-        return self._send_queue.put(data)
-
-    async def recv(self):
-        """
-        Receives a packet of data through this connection mode.
-
-        This method returns a coroutine.
-        """
-        while self._connected:
-            result = await self._recv_queue.get()
-            if result:  # None = sentinel value = keep trying
-                return result
-
-        raise ConnectionError('Not connected')
-
-    async def _send_loop(self):
-        """
-        This loop is constantly popping items off the queue to send them.
-        """
-        try:
-            while self._connected:
-                self._send(await self._send_queue.get())
-                await self._writer.drain()
-        except asyncio.CancelledError:
-            pass
-        except Exception as e:
-            if isinstance(e, IOError):
-                self._log.info('The server closed the connection while sending')
-            else:
-                self._log.exception('Unexpected exception in the send loop')
-
-            await self.disconnect()
-
-    async def _recv_loop(self):
-        """
-        This loop is constantly putting items on the queue as they're read.
-        """
-        while self._connected:
-            try:
-                data = await self._recv()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                if isinstance(e, (IOError, asyncio.IncompleteReadError)):
-                    msg = 'The server closed the connection'
-                    self._log.info(msg)
-                elif isinstance(e, InvalidChecksumError):
-                    msg = 'The server response had an invalid checksum'
-                    self._log.info(msg)
-                else:
-                    msg = 'Unexpected exception in the receive loop'
-                    self._log.exception(msg)
-
-                await self.disconnect()
-
-                # Add a sentinel value to unstuck recv
-                if self._recv_queue.empty():
-                    self._recv_queue.put_nowait(None)
-
-                break
-
-            try:
-                await self._recv_queue.put(data)
-            except asyncio.CancelledError:
-                break
-
-    def _init_conn(self):
-        """
-        This method will be called after `connect` is called.
-        After this method finishes, the writer will be drained.
-
-        Subclasses should make use of this if they need to send
-        data to Telegram to indicate which connection mode will
-        be used.
-        """
-        if self._codec.tag:
-            self._writer.write(self._codec.tag)
-
-    def _send(self, data):
-        self._writer.write(self._codec.encode_packet(data))
-
-    async def _recv(self):
-        return await self._codec.read_packet(self._reader)
-
-    def __str__(self):
-        return '{}:{}/{}'.format(
-            self._ip, self._port,
-            self.__class__.__name__.replace('Connection', '')
-        )
-
-
-class ObfuscatedConnection(Connection):
-    """
-    Base class for "obfuscated" connections ("obfuscated2", "mtproto proxy")
-    """
-    """
-    This attribute should be redefined by subclasses
-    """
-    obfuscated_io = None
-
-    def _init_conn(self):
-        self._obfuscation = self.obfuscated_io(self)
-        self._writer.write(self._obfuscation.header)
-
-    def _send(self, data):
-        self._obfuscation.write(self._codec.encode_packet(data))
-
-    async def _recv(self):
-        return await self._codec.read_packet(self._obfuscation)
-
-
-class PacketCodec(abc.ABC):
-    """
-    Base class for packet codecs
-    """
-
-    """
-    This attribute should be re-defined by subclass to define if some
-    "magic bytes" should be sent to server right after conection is made to
-    signal which protocol will be used
-    """
-    tag = None
-
-    def __init__(self, connection):
-        """
-        Codec is created when connection is just made.
-        """
-        self._conn = connection
-
-    @abc.abstractmethod
-    def encode_packet(self, data):
-        """
-        Encodes single packet and returns encoded bytes.
-        """
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    async def read_packet(self, reader):
-        """
-        Reads single packet from `reader` object that should have
-        `readexactly(n)` method.
-        """
-        raise NotImplementedError

telethon/network/connection/http.py

@@ -1,39 +0,0 @@
-import asyncio
-
-from .connection import Connection, PacketCodec
-
-
-SSL_PORT = 443
-
-
-class HttpPacketCodec(PacketCodec):
-    tag = None
-    obfuscate_tag = None
-
-    def encode_packet(self, data):
-        return ('POST /api HTTP/1.1\r\n'
-                'Host: {}:{}\r\n'
-                'Content-Type: application/x-www-form-urlencoded\r\n'
-                'Connection: keep-alive\r\n'
-                'Keep-Alive: timeout=100000, max=10000000\r\n'
-                'Content-Length: {}\r\n\r\n'
-                .format(self._conn._ip, self._conn._port, len(data))
-                .encode('ascii') + data)
-
-    async def read_packet(self, reader):
-        while True:
-            line = await reader.readline()
-            if not line or line[-1] != b'\n':
-                raise asyncio.IncompleteReadError(line, None)
-
-            if line.lower().startswith(b'content-length: '):
-                await reader.readexactly(2)
-                length = int(line[16:-2])
-                return await reader.readexactly(length)
-
-
-class ConnectionHttp(Connection):
-    packet_codec = HttpPacketCodec
-
-    async def connect(self, timeout=None, ssl=None):
-        await super().connect(timeout=timeout, ssl=self._port == SSL_PORT)

telethon/network/connection/tcpabridged.py

@@ -1,33 +0,0 @@
-import struct
-
-from .connection import Connection, PacketCodec
-
-
-class AbridgedPacketCodec(PacketCodec):
-    tag = b'\xef'
-    obfuscate_tag = b'\xef\xef\xef\xef'
-
-    def encode_packet(self, data):
-        length = len(data) >> 2
-        if length < 127:
-            length = struct.pack('B', length)
-        else:
-            length = b'\x7f' + int.to_bytes(length, 3, 'little')
-        return length + data
-
-    async def read_packet(self, reader):
-        length = struct.unpack('<B', await reader.readexactly(1))[0]
-        if length >= 127:
-            length = struct.unpack(
-                '<i', await reader.readexactly(3) + b'\0')[0]
-
-        return await reader.readexactly(length << 2)
-
-
-class ConnectionTcpAbridged(Connection):
-    """
-    This is the mode with the lowest overhead, as it will
-    only require 1 byte if the packet length is less than
-    508 bytes (127 << 2, which is very common).
-    """
-    packet_codec = AbridgedPacketCodec

telethon/network/connection/tcpintermediate.py

@@ -1,46 +0,0 @@
-import struct
-import random
-import os
-
-from .connection import Connection, PacketCodec
-
-
-class IntermediatePacketCodec(PacketCodec):
-    tag = b'\xee\xee\xee\xee'
-    obfuscate_tag = tag
-
-    def encode_packet(self, data):
-        return struct.pack('<i', len(data)) + data
-
-    async def read_packet(self, reader):
-        length = struct.unpack('<i', await reader.readexactly(4))[0]
-        return await reader.readexactly(length)
-
-
-class RandomizedIntermediatePacketCodec(IntermediatePacketCodec):
-    """
-    Data packets are aligned to 4bytes. This codec adds random bytes of size
-    from 0 to 3 bytes, which are ignored by decoder.
-    """
-    tag = None
-    obfuscate_tag = b'\xdd\xdd\xdd\xdd'
-
-    def encode_packet(self, data):
-        pad_size = random.randint(0, 3)
-        padding = os.urandom(pad_size)
-        return super().encode_packet(data + padding)
-
-    async def read_packet(self, reader):
-        packet_with_padding = await super().read_packet(reader)
-        pad_size = len(packet_with_padding) % 4
-        if pad_size > 0:
-            return packet_with_padding[:-pad_size]
-        return packet_with_padding
-
-
-class ConnectionTcpIntermediate(Connection):
-    """
-    Intermediate mode between `ConnectionTcpFull` and `ConnectionTcpAbridged`.
-    Always sends 4 extra bytes for the packet length.
-    """
-    packet_codec = IntermediatePacketCodec

telethon/network/connection/tcpmtproxy.py

@@ -126,6 +126,7 @@ class TcpMTProxy(ObfuscatedConnection):
 
     @staticmethod
     def address_info(proxy_info):
+        raise NotImplementedError('New proxy format is not implemented')
         if proxy_info is None:
             raise ValueError("No proxy info specified for MTProxy connection")
         return proxy_info[:2]

telethon/network/mtprotosender.py

@@ -122,7 +122,8 @@ class MTProtoSender:
         await self._connect()
         self._user_connected = True
 
-    def is_connected(self):
+    @property
+    def connected(self):
         return self._user_connected
 
     async def disconnect(self):

telethon/sessions/abstract.py

@@ -144,24 +144,3 @@ class Session(ABC):
         to use a cached username to avoid extra RPC).
         """
         raise NotImplementedError
-
-    @abstractmethod
-    def cache_file(self, md5_digest, file_size, instance):
-        """
-        Caches the given file information persistently, so that it
-        doesn't need to be re-uploaded in case the file is used again.
-
-        The ``instance`` will be either an ``InputPhoto`` or ``InputDocument``,
-        both with an ``.id`` and ``.access_hash`` attributes.
-        """
-        raise NotImplementedError
-
-    @abstractmethod
-    def get_file(self, md5_digest, file_size, cls):
-        """
-        Returns an instance of ``cls`` if the ``md5_digest`` and ``file_size``
-        match an existing saved record. The class will either be an
-        ``InputPhoto`` or ``InputDocument``, both with two parameters
-        ``id`` and ``access_hash`` in that order.
-        """
-        raise NotImplementedError

telethon/sessions/memory.py

@@ -1,29 +1,12 @@
-from enum import Enum
-
 from .abstract import Session
 from .. import utils
 from ..tl import TLObject
 from ..tl.types import (
     PeerUser, PeerChat, PeerChannel,
-    InputPeerUser, InputPeerChat, InputPeerChannel,
-    InputPhoto, InputDocument
+    InputPeerUser, InputPeerChat, InputPeerChannel
 )
 
 
-class _SentFileType(Enum):
-    DOCUMENT = 0
-    PHOTO = 1
-
-    @staticmethod
-    def from_type(cls):
-        if cls == InputDocument:
-            return _SentFileType.DOCUMENT
-        elif cls == InputPhoto:
-            return _SentFileType.PHOTO
-        else:
-            raise ValueError('The cls must be either InputDocument/InputPhoto')
-
-
 class MemorySession(Session):
     def __init__(self):
         super().__init__()
@@ -34,7 +17,6 @@ class MemorySession(Session):
         self._auth_key = None
         self._takeout_id = None
 
-        self._files = {}
         self._entities = set()
         self._update_states = {}
 
@@ -228,17 +210,3 @@ class MemorySession(Session):
                 return InputPeerChannel(entity_id, entity_hash)
         else:
             raise ValueError('Could not find input entity with key ', key)
-
-    def cache_file(self, md5_digest, file_size, instance):
-        if not isinstance(instance, (InputDocument, InputPhoto)):
-            raise TypeError('Cannot cache %s instance' % type(instance))
-        key = (md5_digest, file_size, _SentFileType.from_type(type(instance)))
-        value = (instance.id, instance.access_hash)
-        self._files[key] = value
-
-    def get_file(self, md5_digest, file_size, cls):
-        key = (md5_digest, file_size, _SentFileType.from_type(cls))
-        try:
-            return cls(*self._files[key])
-        except KeyError:
-            return None

telethon/sessions/sqlite.py

@@ -2,11 +2,11 @@ import datetime
 import os
 
 from telethon.tl import types
-from .memory import MemorySession, _SentFileType
+from .memory import MemorySession
 from .. import utils
 from ..crypto import AuthKey
 from ..tl.types import (
-    InputPhoto, InputDocument, PeerUser, PeerChat, PeerChannel
+    PeerUser, PeerChat, PeerChannel
 )
 
 try:
@@ -17,7 +17,7 @@ except ImportError as e:
     sqlite3_err = type(e)
 
 EXTENSION = '.session'
-CURRENT_VERSION = 5  # database version
+CURRENT_VERSION = 6  # database version
 
 
 class SQLiteSession(MemorySession):
@@ -87,15 +87,6 @@ class SQLiteSession(MemorySession):
                     name text
                 )"""
                 ,
-                """sent_files (
-                    md5_digest blob,
-                    file_size integer,
-                    type integer,
-                    id integer,
-                    hash integer,
-                    primary key(md5_digest, file_size, type)
-                )"""
-                ,
                 """update_state (
                     id integer primary key,
                     pts integer,
@@ -143,6 +134,9 @@ class SQLiteSession(MemorySession):
         if old == 4:
             old += 1
             c.execute("alter table sessions add column takeout_id integer")
+        if old == 5:
+            old += 1
+            c.execute('drop table sent_files')
         c.close()
 
     @staticmethod
@@ -300,26 +294,3 @@ class SQLiteSession(MemorySession):
                 utils.get_peer_id(PeerChat(id)),
                 utils.get_peer_id(PeerChannel(id))
             )
-
-    # File processing
-
-    def get_file(self, md5_digest, file_size, cls):
-        row = self._execute(
-            'select id, hash from sent_files '
-            'where md5_digest = ? and file_size = ? and type = ?',
-            md5_digest, file_size, _SentFileType.from_type(cls).value
-        )
-        if row:
-            # Both allowed classes have (id, access_hash) as parameters
-            return cls(row[0], row[1])
-
-    def cache_file(self, md5_digest, file_size, instance):
-        if not isinstance(instance, (InputDocument, InputPhoto)):
-            raise TypeError('Cannot cache %s instance' % type(instance))
-
-        self._execute(
-            'insert or replace into sent_files values (?,?,?,?,?)',
-            md5_digest, file_size,
-            _SentFileType.from_type(type(instance)).value,
-            instance.id, instance.access_hash
-        )

telethon_examples/gui.py

@@ -228,7 +228,7 @@ class App(tkinter.Tk):
         """
         Sends a message. Does nothing if the client is not connected.
         """
-        if not self.cl.is_connected():
+        if not self.cl.connected:
             return
 
         # The user needs to configure a chat where the message should be sent.

telethon_examples/interactive_telegram_client.py

@@ -116,9 +116,8 @@ class InteractiveTelegramClient(TelegramClient):
         try:
             loop.run_until_complete(self.connect())
         except IOError:
-            # We handle IOError and not ConnectionError because
-            # PySocks' errors do not subclass ConnectionError
-            # (so this will work with and without proxies).
+            # To avoid issues in the future, we except the most
+            # generic IOError as possible (instead of ConnectionError)
             print('Initial connection failed. Retrying...')
             loop.run_until_complete(self.connect())
 

telethon_examples/print_messages.py

@@ -22,7 +22,7 @@ def get_env(name, message, cast=str):
 session = os.environ.get('TG_SESSION', 'printer')
 api_id = get_env('TG_API_ID', 'Enter your API ID: ', int)
 api_hash = get_env('TG_API_HASH', 'Enter your API hash: ')
-proxy = None  # https://github.com/Anorov/PySocks
+proxy = None  # https://docs.telethon.dev/en/latest/basic/signing-in.html#signing-in-behind-a-proxy
 
 # Create and start the client so we can make requests (we don't here)
 client = TelegramClient(session, api_id, api_hash, proxy=proxy).start()

telethon_examples/print_updates.py

@@ -27,7 +27,7 @@ def get_env(name, message, cast=str):
 session = os.environ.get('TG_SESSION', 'printer')
 api_id = get_env('TG_API_ID', 'Enter your API ID: ', int)
 api_hash = get_env('TG_API_HASH', 'Enter your API hash: ')
-proxy = None  # https://github.com/Anorov/PySocks
+proxy = None  # https://docs.telethon.dev/en/latest/basic/signing-in.html#signing-in-behind-a-proxy
 
 
 # This is our update handler. It is called when a new update arrives.