示例代碼：

import requestsimport osdef downlaod(url, file_path): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:68.0) Gecko/20100101 Firefox/68.0' } r = requests.get(url=url, headers=headers) with open(file_path, 'wb') as f: f.write(r.content) f.flush()2、 使用流式請求，requests.get方法的stream

默認情況下是stream的值為false，它會立即開始下載文件并存放到內存當中，倘若文件過大就會導致內存不足的情況，程序就會報錯。當把get函數的stream參數設置成True時，它不會立即開始下載，當你使用iter_content或iter_lines遍歷內容或訪問內容屬性時才開始下載，需要注意一點：文件沒有下載之前，它也需要保持連接。

iter_content：一塊一塊的遍歷要下載的內容iter_lines：一行一行的遍歷要下載的內容

使用上面兩個函數下載大文件可以防止占用過多的內存，因為每次只下載小部分數據。

示例代碼:

由于request的請求是阻塞式的，所以要用aiohttp模塊來發起請求。

示例代碼：

import aiohttpimport asyncioimport osasync def handler(url, file_path): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:68.0) Gecko/20100101 Firefox/68.0' } async with aiohttp.ClientSession() as session: r = await session.get(url=url, headers=headers) with open(file_path, 'wb') as f: f.write(await r.read()) f.flush() os.fsync(f.fileno())loop = asyncio.get_event_loop()loop.run_until_complete(handler(url, file_path))4、 異步拆分下載文件

上面用的是一個協程下載一個文件，下面的方法是將文件分成幾部分，每個部分用一個協程下載，最后再寫入文件。

下面這個例子用的是流式寫入，即把內容寫入到磁盤里面。

import aiohttpimport asyncioimport timeimport osasync def consumer(queue): option = await queue.get() start = option['start'] end = option['end'] url = option['url'] filename = option['filename'] i = option['i'] print(f'第{i}個任務開始運行') async with aiohttp.ClientSession() as session: headers = {'Range': f'bytes={start}-{end}'} r = await session.get(url=url, headers=headers) with open(filename, 'rb+') as f: f.seek(start) while True:chunk = await r.content.read(end - start)if not chunk: breakf.write(chunk)f.flush()os.fsync(f.fileno())print(f'第{i}個任務正在寫入中ing') queue.task_done() print(f'第{i}個任務寫入成功')async def producer(url, headers, filename, queue, coro_num): async with aiohttp.ClientSession() as session: resp = await session.head(url=url, headers=headers) file_size = int(resp.headers['content-length']) # 創建一個文件 with open(filename, 'wb') as f: pass part = file_size // coro_num for i in range(coro_num): start = part * i if i == coro_num - 1:end = file_size else:end = start + part info = {'start': start,'end': end,'url': url,'filename': filename,'i': i, } queue.put_nowait(info)async def main(): # 需要填的有url，filename，coro_num url = '' filename = '' coro_num = 0 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:68.0) Gecko/20100101 Firefox/68.0' } queue = asyncio.Queue(coro_num) await producer(url, headers, filename, queue, coro_num) task_list = [] for i in range(coro_num): task = asyncio.create_task(consumer(queue)) task_list.append(task) await queue.join() for i in task_list: i.cancel() await asyncio.gather(*task_list)startt = time.time()loop = asyncio.get_event_loop()loop.run_until_complete(main())end = time.time() - starttprint(f'用了{end}秒')5、注意

以上的示例都是介紹思路，程序并不健壯，健壯的程序需要加入錯誤捕獲和錯誤處理。

以上就是python 下載文件的幾種方式分享的詳細內容，更多關于python 下載文件的資料請關注好吧啦網其它相關文章！