✨ Add URLInputFile

aiogram/api/types/__init__.py

@@ -41,7 +41,7 @@ from .inline_query_result_venue import InlineQueryResultVenue
 from .inline_query_result_video import InlineQueryResultVideo
 from .inline_query_result_voice import InlineQueryResultVoice
 from .input_contact_message_content import InputContactMessageContent
-from .input_file import BufferedInputFile, FSInputFile, InputFile
+from .input_file import BufferedInputFile, FSInputFile, InputFile, URLInputFile
 from .input_location_message_content import InputLocationMessageContent
 from .input_media import InputMedia
 from .input_media_animation import InputMediaAnimation
@@ -99,6 +99,7 @@ __all__ = (
     "TelegramObject",
     "BufferedInputFile",
     "FSInputFile",
+    "URLInputFile",
     "Update",
     "WebhookInfo",
     "User",

aiogram/api/types/input_file.py

@@ -6,7 +6,7 @@ from abc import ABC, abstractmethod
 from pathlib import Path
 from typing import AsyncGenerator, Optional, Union
 
-import aiofiles as aiofiles
+import aiofiles
 
 DEFAULT_CHUNK_SIZE = 64 * 1024  # 64 kb
 
@@ -82,3 +82,28 @@ class FSInputFile(InputFile):
             while chunk:
                 yield chunk
                 chunk = await f.read(chunk_size)
+
+
+class URLInputFile(InputFile):
+    def __init__(
+        self,
+        url: str,
+        filename: Optional[str] = None,
+        chunk_size: int = DEFAULT_CHUNK_SIZE,
+        timeout: int = 30,
+    ):
+        super().__init__(filename=filename, chunk_size=chunk_size)
+
+        self.url = url
+        self.timeout = timeout
+
+    async def read(self, chunk_size: int) -> AsyncGenerator[bytes, None]:
+        from aiogram.api.client.bot import Bot
+
+        bot = Bot.get_current(no_error=False)
+        stream = bot.session.stream_content(
+            url=self.url, timeout=self.timeout, chunk_size=self.chunk_size
+        )
+
+        async for chunk in stream:
+            yield chunk

docs/api/sending_files.md

@@ -7,6 +7,7 @@ But if you need to upload new file just use subclasses of [InputFile](./types/in
 
 - `#!python3 FSInputFile` - [uploading from file system](#upload-from-file-system)
 - `#!python3 BufferedInputFile` - [uploading from buffer](#upload-from-buffer)
+- `#!python3 URLInputFile` - [uploading from URL](#upload-from-url)
 
 !!! warning "Be respectful with Telegram"
     Instances of `InputFile` is reusable. That's mean you can create instance of InputFile and sent this file multiple times but Telegram is not recommend to do that and when you upload file once just save their `file_id` and use it in next times.
@@ -65,3 +66,26 @@ file = BufferedInputFile.from_file("file.txt")
 | `path` | `#!python3 Union[str, Path]` | File path |
 | `filename` | `#!python3 Optional[str]` | Custom filename to be presented to Telegram |
 | `chunk_size` | `#!python3 int` | File chunks size (Default: `64 kb`) |
+
+## Upload from url
+
+If you need to upload a file from another server, but the direct link is bound to your server's IP, or you want to bypass native [upload limits](https://core.telegram.org/bots/api#sending-files) by URL, you can use [URLInputFile](#urlinputfile).
+
+Import wrapper:
+
+```python3
+from aiogram.types import URLInputFile
+```
+
+And then you can use it:
+```python3
+image = URLInputFile("https://www.python.org/static/community_logos/python-powered-h-140x182.png", filename="logo.png")
+```
+
+### URLInputFile(...)
+|Argument|Type|Description|
+|---|---|---|
+| `url` | `#!python3 str` | URL |
+| `filename` | `#!python3 Optional[str]` | Custom filename to be presented to Telegram |
+| `chunk_size` | `#!python3 int` | File chunks size (Default: `64 kb`) |
+| `timeout` | `#!python3 int` | Total timeout in seconds (Default: `30`) |

tests/test_api/test_types/test_input_file.py

@@ -1,8 +1,10 @@
 from typing import AsyncIterable
 
 import pytest
+from aresponses import ResponsesMockServer
 
-from aiogram.api.types import BufferedInputFile, FSInputFile, InputFile
+from aiogram.api.client.bot import Bot
+from aiogram.api.types import BufferedInputFile, FSInputFile, InputFile, URLInputFile
 
 
 class TestInputFile:
@@ -70,3 +72,21 @@ class TestInputFile:
             assert chunk_size == 1
             size += chunk_size
         assert size > 0
+
+    @pytest.mark.asyncio
+    async def test_uri_input_file(self, aresponses: ResponsesMockServer):
+        aresponses.add(
+            aresponses.ANY, aresponses.ANY, "get", aresponses.Response(status=200, body=b"\f" * 10)
+        )
+
+        Bot.set_current(Bot("42:TEST"))
+
+        file = URLInputFile("https://test.org/", chunk_size=1)
+
+        size = 0
+        async for chunk in file:
+            assert chunk == b"\f"
+            chunk_size = len(chunk)
+            assert chunk_size == 1
+            size += chunk_size
+        assert size == 10