add Bot.download_file aliases ability to save files to a directory and automatically create directories (#694)

* add destination_dir and make_dirs parameters to bot download aliases

* add the ability to save files to a directory with path completion based on file_path,
* add an option to automatically create directories in the file path

* Downloadable mixin uses directory creation parameter in bot methods
This commit is contained in:
darksidecat 2021-11-07 01:28:12 +02:00 committed by GitHub
parent 204a2a1ec0
commit b98ec3efad
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 127 additions and 17 deletions

View file

@ -1,6 +1,8 @@
import asyncio
import contextlib
import io
import os
import pathlib
import ssl
import typing
import warnings
@ -208,28 +210,48 @@ class BaseBot:
return await api.make_request(self.session, self.server, self.__token, method, data, files,
proxy=self.proxy, proxy_auth=self.proxy_auth, timeout=self.timeout, **kwargs)
async def download_file(self, file_path: base.String,
destination: Optional[base.InputFile] = None,
timeout: Optional[base.Integer] = sentinel,
chunk_size: Optional[base.Integer] = 65536,
seek: Optional[base.Boolean] = True) -> Union[io.BytesIO, io.FileIO]:
async def download_file(
self,
file_path: base.String,
destination: Optional[Union[base.InputFile, pathlib.Path]] = None,
timeout: Optional[base.Integer] = sentinel,
chunk_size: Optional[base.Integer] = 65536,
seek: Optional[base.Boolean] = True,
destination_dir: Optional[Union[str, pathlib.Path]] = None,
make_dirs: Optional[base.Boolean] = True,
) -> Union[io.BytesIO, io.FileIO]:
"""
Download file by file_path to destination
Download file by file_path to destination file or directory
if You want to automatically create destination (:class:`io.BytesIO`) use default
value of destination and handle result of this method.
At most one of these parameters can be used: :param destination:, :param destination_dir:
:param file_path: file path on telegram server (You can get it from :obj:`aiogram.types.File`)
:type file_path: :obj:`str`
:param destination: filename or instance of :class:`io.IOBase`. For e. g. :class:`io.BytesIO`
:param timeout: Integer
:param chunk_size: Integer
:param seek: Boolean - go to start of file when downloading is finished.
:param destination_dir: directory for saving files
:param make_dirs: Make dirs if not exist
:return: destination
"""
if destination is None:
if destination and destination_dir:
raise ValueError(
"Use only one of the parameters:destination or destination_dir."
)
if destination is None and destination_dir is None:
destination = io.BytesIO()
elif destination_dir:
destination = os.path.join(destination_dir, file_path)
if make_dirs and not isinstance(destination, io.IOBase) and os.path.dirname(destination):
os.makedirs(os.path.dirname(destination), exist_ok=True)
url = self.get_file_url(file_path)
dest = destination if isinstance(destination, io.IOBase) else open(destination, 'wb')

View file

@ -1,6 +1,7 @@
from __future__ import annotations
import datetime
import pathlib
import typing
import warnings
@ -43,25 +44,37 @@ class Bot(BaseBot, DataMixin, ContextInstanceMixin):
if hasattr(self, '_me'):
delattr(self, '_me')
async def download_file_by_id(self, file_id: base.String, destination=None,
timeout: base.Integer = 30, chunk_size: base.Integer = 65536,
seek: base.Boolean = True):
async def download_file_by_id(
self,
file_id: base.String,
destination: typing.Optional[base.InputFile, pathlib.Path] = None,
timeout: base.Integer = 30,
chunk_size: base.Integer = 65536,
seek: base.Boolean = True,
destination_dir: typing.Optional[typing.Union[str, pathlib.Path]] = None,
make_dirs: typing.Optional[base.Boolean] = True,
):
"""
Download file by file_id to destination
Download file by file_id to destination file or directory
if You want to automatically create destination (:class:`io.BytesIO`) use default
value of destination and handle result of this method.
At most one of these parameters can be used: :param destination:, :param destination_dir:
:param file_id: str
:param destination: filename or instance of :class:`io.IOBase`. For e. g. :class:`io.BytesIO`
:param timeout: int
:param chunk_size: int
:param seek: bool - go to start of file when downloading is finished
:param destination_dir: directory for saving files
:param make_dirs: Make dirs if not exist
:return: destination
"""
file = await self.get_file(file_id)
return await self.download_file(file_path=file.file_path, destination=destination,
timeout=timeout, chunk_size=chunk_size, seek=seek)
timeout=timeout, chunk_size=chunk_size, seek=seek,
destination_dir=destination_dir, make_dirs=make_dirs)
# === Getting updates ===
# https://core.telegram.org/bots/api#getting-updates

View file

@ -49,7 +49,6 @@ class Downloadable:
destination,
destination_dir,
destination_file,
make_dirs
)
return await self.bot.download_file(
@ -58,9 +57,10 @@ class Downloadable:
timeout=timeout,
chunk_size=chunk_size,
seek=seek,
make_dirs=make_dirs
)
async def _prepare_destination(self, dest, destination_dir, destination_file, make_dirs):
async def _prepare_destination(self, dest, destination_dir, destination_file):
file = await self.get_file()
if not(any((dest, destination_dir, destination_file))):
@ -87,9 +87,6 @@ class Downloadable:
else:
raise TypeError("destination_file must be str, pathlib.Path or io.IOBase type")
if make_dirs and os.path.dirname(destination):
os.makedirs(os.path.dirname(destination), exist_ok=True)
return file, destination
async def get_file(self):