downloader.py 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158
  1. import asyncio
  2. import os
  3. from typing import Optional
  4. import aiofiles
  5. import aiohttp
  6. class DownloadJob:
  7. """
  8. Download Job
  9. :param file_url: url where the file is located
  10. :param session: aiohttp session to be used on the job
  11. :param file_name: name to be used on the file. Defaults to the last part of the url
  12. :param save_path: dir where the file should be saved. Defaults to the current dir
  13. """
  14. def __init__(
  15. self,
  16. session: aiohttp.ClientSession,
  17. file_url: str,
  18. save_path: Optional[str] = None,
  19. chunk_size: Optional[int] = 1024,
  20. ):
  21. self.file_url = file_url
  22. self._session = session
  23. self._chunk_size = chunk_size
  24. self.file_name = file_url.split("/")[~0][0:230]
  25. self.file_path = (
  26. os.path.join(save_path, self.file_name)
  27. if save_path
  28. else self.file_name
  29. )
  30. self.completed = False
  31. self.progress = 0
  32. self.size = 0
  33. async def get_size(self) -> int:
  34. """
  35. Gets the content-length of the file from the file url.
  36. :return: the files size in bytes
  37. """
  38. if not self.size:
  39. async with self._session.get(self.file_url) as resp:
  40. if 200 <= resp.status < 300:
  41. self.size = int(resp.headers["Content-Length"])
  42. else:
  43. raise aiohttp.errors.HttpProcessingError(
  44. message=f"There was a problem processing {self.file_url}",
  45. code=resp.status,
  46. )
  47. return self.size
  48. def _downloaded(self, chunk_size):
  49. """
  50. Method to be called when a chunk of the file is downloaded. It updates the
  51. progress, adds the size to the _advance variable and checks if the download
  52. is completed
  53. """
  54. self.progress += chunk_size
  55. async def download(self):
  56. """
  57. Downloads the file from the given url to a file in the given path.
  58. """
  59. async with self._session.get(self.file_url) as resp:
  60. # Checkning the response code
  61. if 200 <= resp.status < 300:
  62. # Saving the data to the file chunk by chunk.
  63. async with aiofiles.open(
  64. self.file_path, "wb"
  65. ) as file:
  66. # Downloading the file using the aiohttp.StreamReader
  67. async for data in resp.content.iter_chunked(
  68. self._chunk_size
  69. ):
  70. await file.write(data)
  71. self._downloaded(self._chunk_size)
  72. self.completed = True
  73. return self
  74. else:
  75. raise aiohttp.errors.HttpProcessingError(
  76. message=f"There was a problem processing {self.file_url}",
  77. code=resp.status,
  78. )
  79. class Handler:
  80. """
  81. Top level interface with the downloader. It creates the download jobs and handles them.
  82. :param loop: asyncio loop. if not provided asyncio.get_event_loop() will be used
  83. :param session: aiohttp session to be used on all downloads. If not provided it will be
  84. created
  85. :param chunk_size: chunk bytes sizes to get from the file source. Defaults to 1024 bytes
  86. """
  87. def __init__(
  88. self,
  89. loop: Optional[asyncio.BaseEventLoop] = None,
  90. session: Optional[aiohttp.ClientSession] = None,
  91. chunk_size: Optional[int] = 1024,
  92. ):
  93. self._loop = loop or asyncio.get_event_loop()
  94. self._session = session or aiohttp.ClientSession(
  95. loop=self._loop
  96. )
  97. self._chunk_size = chunk_size
  98. def _job_factory(
  99. self, file_url: str, save_path: Optional[str] = None
  100. ) -> DownloadJob:
  101. """
  102. Shortcut for creating a download job. It adds the session and the chunk size.
  103. :param file_url: url where the file is located
  104. :param save_path: save path for the download
  105. :return:
  106. """
  107. return DownloadJob(
  108. self._session, file_url, save_path, self._chunk_size
  109. )
  110. async def download(
  111. self, url: str, save_path: Optional[str] = None
  112. ) -> DownloadJob:
  113. """
  114. Downloads a bulk of files from the given list of urls to the given path.
  115. :param files_url: list of urls where the files are located
  116. :param save_path: path to be used for saving the files. Defaults to the current dir
  117. """
  118. job = self._job_factory(url, save_path=save_path)
  119. task = asyncio.ensure_future(job.download())
  120. await task
  121. file_name = url.split("/")[-1]
  122. file_name = (
  123. file_name[0:230] if len(file_name) > 230 else file_name
  124. )
  125. path = (
  126. os.getcwd() + "/" + file_name
  127. if not save_path
  128. else save_path
  129. )
  130. return path