本文专门用来记录一下python中一些好用方法/库,可以在日常使用中提高效率。
进度条 在爬虫和机器学习等工作中,可能需要有一个进度条能够反馈当前程序运行速度或者进度,可以考虑用以下方法实现:
tqdm 老朋友tqdm,就是丑了点。
安装
使用 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 import timefrom tqdm import tqdmfor i in tqdm(range (9 )): time.sleep(0.1 ) >>> 100 %|██████████| 10 /10 [00 :01<00 :00 , 9.79 it/s]import timefrom tqdm import trangefor i in trange(10 ): time.sleep(0.1 ) >>> 100 %|██████████| 10 /10 [00 :01<00 :00 , 9.79 it/s]import timefrom tqdm import tqdmpbar = tqdm([1 ,2 ,3 ,4 ,5 ,6 ,7 ,8 ,9 ,10 ]) for char in pbar: pbar.set_description("Processing %s" % char) time.sleep(0.1 ) >>> Processing 10 : 100 %|██████████| 10 /10 [00 :01<00 :00 , 9.49 it/s] import timefrom tqdm import tqdmwith tqdm(total=10 ) as pbar: for i in range (10 ): pbar.update(1 ) time.sleep(0.1 ) >>> 100 %|██████████| 10 /10 [00 :00 <00 :00 , 10.10 it/s]import timefrom tqdm import tqdmpbar = tqdm(total=10 ) for i in range (10 ): pbar.update(1 ) time.sleep(0.1 ) pbar.close() >>> 100 %|██████████| 10 /10 [00 :00 <00 :00 , 10.10 it/s]
在Spyder下正常了,然而在命令窗口有问题。
Rich 非常炫酷的包,当然这里仅仅用它的进度条。(由@kotori-y投稿)
安装
使用 基本用法:
1 2 3 4 from rich.progress import trackfor step in track(range (100 )): do_step(step)
按步更新:
1 2 3 4 5 6 from rich.progress import Progresswith Progress() as progress: task = progress.add_task("[red]Downloading..." , total=total) progress.update(task, advance=1 )
存在问题在于,不能在cmder或者spyder下正常显示,有请Windows Terminal(逃)。
ShowProcess类 在网上找到别人写的一个方法如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 class ShowProcess (): i = 0 max_steps = 0 max_arrow = 50 def __init__ (self, max_steps ): self .max_steps = max_steps self .i = 0 def show_process (self, i=None ): if i is not None : self .i = i else : self .i += 1 num_arrow = int (self .i * self .max_arrow / self .max_steps) num_line = self .max_arrow - num_arrow percent = self .i * 100.0 / self .max_steps process_bar = '[' + '>' * num_arrow + '-' * num_line + ']' \ + '%.2f' % percent + '%' print ('\r' ,process_bar,end='' ,flush=True ) def close (self, words='done' ): print ('' ) print (words) self .i = 0
使用示例:
1 2 3 4 5 6 7 8 9 10 11 12 13 if __name__=='__main__' : max_steps = 100 process_bar = ShowProcess(max_steps) for i in range (max_steps): process_bar.show_process() time.sleep(0.05 ) process_bar.close() >>> [>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>]100.00 %>>> done
在命令窗口下使用正常,但是在IDLE和Spyder中显示存在问题,考虑使用其他方法。
重试 进行爬虫的时候,很容易因为网络问题导致失败,这里有2个库可以很轻松地实现这个功能。
retry 安装
使用 只需要在函数定义前加上@retry就行了。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from retry import retry@retry() def make_trouble (): '''重试直到成功''' @retry(ZeroDivisionError, tries=3 , delay=2 ) def make_trouble (): '''出现ZeroDivisionError时重试, 重试3次,每次间隔2秒''' @retry((ValueError, TypeError ), delay=1 , backoff=2 ) def make_trouble (): '''出现ValueError或TypeError时重试, 每次间隔1, 2, 4, 8, ...秒''' @retry((ValueError, TypeError ), delay=1 , backoff=2 , max_delay=4 ) def make_trouble (): '''出现ValueError或TypeError时重试, 每次间隔1, 2, 4, 4, ...秒,最高间隔为4秒''' @retry(ValueError, delay=1 , jitter=1 ) def make_trouble (): '''出现ValueError时重试,每次间隔1, 2, 3, 4, ... 秒'''
Tenacity 使用类似于retry。同样只需要在函数定义前加上@retry就行了。
安装
使用 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from tenacity import retry, retry_if_exception_type, wait_fixed, stop_after_attempt, stop_after_delay,@retry() def make_trouble (): '''重试直到成功''' @retry(retry=retry_if_exception_type(ZeroDivisionError ), wait=wait_fixed(2 ), stop=stop_after_attempt(3 ) ) def make_trouble (): '''出现ZeroDivisionError时重试, 重试3次,每次间隔2秒''' @retry(stop=(stop_after_delay(10 ) | stop_after_attempt(5 ) ) ) def make_trouble (): '''重试10秒或者5次''' @retry(wait=wait_random(min =1 , max =2 ) ) def make_trouble (): '''重试间隔在随机1-2秒''' @retry(wait=wait_chain(*[wait_fixed(3 ) for i in range (3 )] + [wait_fixed(7 ) for i in range (2 )] + [wait_fixed(9 )] ) )def make_trouble (): '''前三次重试每次间隔3秒,接下来2次间隔7秒,之后重试间隔9秒'''
超时 很多任务特别是多线程时,为了防止程序卡死,需要设定一个超时。
func_timeout 由于windows下signal的支持问题,选择使用第三方包,func_timeout就是一个给函数添加超时的包。
安装 1 pip install func_timeout
使用 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 import timefrom func_timeout import func_set_timeout,FunctionTimedOuttry : doitReturnValue = func_timeout(5 , doit, args=('arg1' , 'arg2' )) except FunctionTimedOut: print ( "doit('arg1', 'arg2') could not complete within 5 seconds and was terminated.\n" ) except Exception as e: @func_set_timeout(2 ) def task (): time.sleep(5 ) FunctionTimedOut: Function task (args=()) (kwargs={}) timed out after 2.000000 seconds. from func_timeout.exceptions import FunctionTimedOuttry : task() except FunctionTimedOut: print ('task func_timeout' )
此外它还有一个重试的函数FunctionTimedOut,就不赘述了。
异步 待填坑……