十四、fileinput模块
十五、shutil模块
十六、zipfile模块
十七、getpass模块
十八、bisect模块
十九、logging模块
二十、loguru模块---管理日志
二十一、schedule模块--定时任务
二十二、yagmail模块--自动发邮件
二十三、tenacity--错误自动重试
二十四、schedule定时任务
二十五、APScheduler组件

十四、fileinput模块

===========================================================================

fileinput是python内置的模块，这个fileinput可以帮助我们轻松的实现文件内容的修改功能。fileinput模块中的重要方法如下：

input([files[, inplace[, backup]]) 最重要的方法，用于遍历打开的文件，inplace为True的时候，会将修改写入文件，backup为True的时候会进行备份操作。

filename() 返回当前文件的名字

lineno() 返回当前（累计）的行数。同时处理多个文件时，行数会累计。

filelineno() 返回当前文件的总行数。处理新文件时，行数会重置为1，重新计数。

isfirstline() 检查当前行是否是文件的第一行

isstdin() 检查最后一行是否来自sys.stdin

nextfile() 关闭当前文件，移动到下一个文件（fileinput是可以同时处理多个文件的！）

close()　 关闭整个文件链，结束迭代。

import fileinput

with fileinput.input(files="d:\\1.txt") as f:
    for line in f:
        line = line.rstrip()
        num = fileinput.lineno()
        print("#%d\t%s" % (num, line))

#输出结果
#1  愿圣光与你同在！
#2  
#3  为了部落！
#4  
#5  兽人永不为奴！
#6  
#7  你们这是自寻死路！
#8  
#9  复活吧我的勇士！
#10 
#11 为你而战我的女士！

十五、shutil模块

====================================================================

shutil是python的内置模块，shutil是一个高级的高级的文件、文件夹、压缩包 处理模块，它本质上是调用open方法对文件进行读写。模块相对比较简单，记住几个常用的方法即可。

shutil.copyfileobj(fsrc, fdst[, length])将文件内容拷贝到另一个文件中。　

shutil.copyfile(src, dst)  拷贝文件　

shutil.copymode(src, dst) 仅拷贝权限。内容、组、用户均不变　

shutil.copystat(src, dst)仅拷贝状态的信息，包括：mode bits, atime, mtime, flags

shutil.copy(src, dst)拷贝文件和权限

shutil.copy2(src, dst)拷贝文件和状态信息 

shutil.ignore_patterns(*patterns)

shutil.copytree(src, dst, symlinks=False, ignore=None) 递归的去拷贝文件夹。ignor_patterns是指忽略不拷贝的文件

shutil.copytree('folder1', 'folder2', ignore=shutil.ignore_patterns('*.pyc', 'tmp*'))

shutil.rmtree(path[, ignore_errors[, onerror]])递归的去删除文件

shutil.move(src, dst)递归的去移动文件，它类似mv命令，其实就是重命名。　

shutil.make_archive(base_name, format,...)创建压缩包并返回文件路径，例如：zip、tar

      ● base_name： 压缩包的文件名，也可以是压缩包的路径。是文件名时则保存至当前目录，否则保存至指定路径，

      ● format： 压缩包种类，“zip”, “tar”, “bztar”，“gztar”

      ●  root_dir： 要压缩的文件夹路径（默认当前目录）

      ● owner： 用户，默认当前用户

      ●  group： 组，默认当前组

      ● logger： 用于记录日志，通常是logging.Logger对象

 

十六、zipfile模块

==========================================================

zipfile是python内置模块，当你需要压缩文件的时候，使用这个模块会比较方便。

import zipfile

# 压缩
z = zipfile.ZipFile('laxi.zip', 'w')
z.write('a.log')       
z.write('data.data')             # 可以一个一个添加文件进压缩包
z.close()

# 解压
z = zipfile.ZipFile('laxi.zip', 'r')
z.extractall()                              # 这是一次性将压缩包内的文件全部解压出来
z.close()   

要单独解压压缩包内的某个文件就需要先获得压缩包内的文件名列表。zipfile提供了一个namelist方法。

import zipfile

z = zipfile.ZipFile('laxi.zip', 'r')
ret = z.namelist()
print(ret)
z.close()

运行结果：

['testfile.bak', 'testfile.dat']

通过具体的文件名去解压某个文件。zipfile提供了extract方法。

import zipfile

z = zipfile.ZipFile('laxi.zip', 'r')
z.extract("testfile.bak")
z.close()

 

十七、getpass模块

===============================================================

getpass模块非常简单，它能够让你在输入密码的时候不会在屏幕上显示密码，安全性更高。注意：在pycharm环境里这个模块用不了！

getpass模块只有2个常用方法：getpass和getuser。

import getpass

pwd = getpass.getpass("请输入密码： ")  # 可代替input方法，接收用户输入的密码
print(getpass.getuser())  # getuser方法会返回当前执行程序的用户名

 

十八、logging模块

===========================================================================

python提供logging日志记录模块，以下是logging模块配置参数

logging.basicConfig()函数可更改logging模块默认行为，可用参数有：

filename：用指定的文件名创建FiledHandler，这样日志会被存储在指定的文件中。
filemode：文件打开方式，在指定了filename时使用这个参数，默认值为“a”还可指定为“w”。
format：指定handler使用的日志显示格式。
datefmt：指定日期时间格式。
level：设置rootlogger（后边会讲解具体概念）的日志级别
stream：用指定的stream创建StreamHandler。可以指定输出到sys.stderr,sys.stdout或者文件，默认为sys.stderr。若同时列出了filename和stream两个参数，则stream参数会被忽略。

format参数中可能用到的格式化串：

%(name)s Logger的名字
%(levelno)s 数字形式的日志级别
%(levelname)s 文本形式的日志级别
%(pathname)s 调用日志输出函数的模块的完整路径名，可能没有
%(filename)s 调用日志输出函数的模块的文件名
%(module)s 调用日志输出函数的模块名
%(funcName)s 调用日志输出函数的函数名
%(lineno)d 调用日志输出函数的语句所在的代码行
%(created)f 当前时间，用UNIX标准的表示时间的浮 点数表示
%(relativeCreated)d 输出日志信息时的，自Logger创建以 来的毫秒数
%(asctime)s 字符串形式的当前时间。默认格式是 “2003-07-08 16:49:45,896”。逗号后面的是毫秒
%(thread)d 线程ID。可能没有
%(threadName)s 线程名。可能没有
%(process)d 进程ID。可能没有
%(message)s用户输出的消息

logging简单测试

import logging 

logging.debug('debug message') 
logging.info('info message') 
logging.warning('warning message') 
logging.error('error message') 
logging.critical('critical message'

默认情况下logging模块将日志打印到了标准输出中，且只显示了大于等于WARNING级别的日志（日志级别等级CRITICAL > ERROR > WARNING > INFO > DEBUG），
默认的日志格式为日志级别：Logger名称：用户输出消息。

logging自定义格式输出

import logging

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
                    datefmt='%a, %d %b %Y %H:%M:%S',
                    filename='test.log',
                    filemode='w')

logging.debug('debug message')
logging.info('info message')
logging.warning('warning message')
logging.error('error message')
logging.critical('critical message')

生成test.log文件，内容如下

logging对象配置

import logging

logger = logging.getLogger()
# 创建一个handler，用于写入日志文件
fh = logging.FileHandler('test.log',encoding='utf-8')

# 再创建一个handler，用于输出到控制台
ch = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)
ch.setFormatter(formatter)

logger.addHandler(fh) #logger对象可以添加多个fh和ch对象
logger.addHandler(ch)
logger.debug('logger debug message')
logger.info('logger info message')
logger.warning('logger warning message')
logger.error('logger error message')
logger.critical('logger critical message')

控制台：

 log文件：

logging库提供了多个组件：Logger、Handler、Filter、Formatter。Logger对象提供应用程序可直接使用的接口，Handler发送日志到适当的目的地，Filter提供了过滤日志信息的方法，Formatter指定日志显示格式。另外，可以通过：logger.setLevel(logging.Debug)设置级别,当然，也可以通过fh.setLevel(logging.Debug)单对文件流设置某个级别。

十九、loguru模块---管理日志

安装：pip3 install loguru

loguru默认的输出格式是：时间、级别、模块名、行号、日志信息。直接使用即可，其输出是彩色的，看起来会更加友好。loguru下只有一个logger类，该类有如下方法，其中最常用，最重要的是add方法

 'add', 'bind', 'catch', 'complete', 'configure', 'contextualize', 'critical', 'debug', 'disable', 'enable', 'error', 'exception', 'info', 'level', 'log', 'opt', 'parse', 'patch',
 'remove', 'start', 'stop', 'success', 'trace', 'warning'

add(self, sink, *, level='DEBUG', 
format='{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}', 
filter=None, colorize=None, serialize=False, backtrace=True, diagnose=True, enqueue=False, catch=True, **kwargs)

add 方法

1 sink参数，通过 sink 我们可以传入多种不同的数据结构，给它传了一个 str 字符串路径，他就给我们创建了一个日志文件：

• sink 可以传入一个 file 对象，例如 sys.stderr 或者 open(‘file.log’, ‘w’) 都可以。
• sink 可以直接传入一个 str 字符串或者 pathlib.Path 对象，其实就是代表文件路径的，如果识别到是这种类型，它会自动创建对应路径的日志文件并将日志输出进去。
• sink 可以是一个方法，可以自行定义输出实现。
• sink 可以是一个 logging 模块的 Handler，比如 FileHandler、StreamHandler 等等。
  sink 还可以是一个自定义的类。

2  format、filter、level参数，这些是用来规定输出的格式：

logger.add('runtime.log', format="{time} {level} {message}", filter="my_module", level="INFO")

3 rotation 参数，可设置日志文件的大小/时间等

from loguru import logger

logger.add('runtime_{time}.log', rotation="500 MB")#每500MB 存储一个文件，log文件过大就会新创建一个log文件。time占位符使系统自动生成一个文件名包含时间的log文件。
logger.add('runtime_{time}.log', rotation='00:00') #这样就可以实现每天 0 点新创建一个 log 文件输出了。
logger.add('runtime_{time}.log', rotation='1 week') #配置 log 文件的循环时间，每隔一周创建一个 log 文件

4 retention 参数，可以配置日志的最长保留时间。比如我们想要设置日志文件最长保留 10 天，这样 log 文件里面就会保留最新 10 天的 log：

from loguru import logger
logger.add('runtime.log', retention='10 days')

5 compression 参数，用来配置文件的压缩格式，比如使用 zip 文件格式保存：

from loguru import logger
logger.add('runtime.log', compression='zip')

6 删除 sink，添加 sink 之后我们也可以对其进行删除，删除的时候根据刚刚 add 方法返回的 id 进行删除：

from loguru import logger

trace = logger.add('my_log.log')
logger.debug('调试信息')
logger.remove(trace)
logger.debug('另一个调试信息')

终端显示的内容：

日志文件 my_log.log 内容如下：

7 字符串格式化

loguru 在输出 log 的时候还提供了非常友好的字符串格式化功能：

from loguru import logger

logger.info('If you are using Python {}, prefer {feature} of course!', 3.6, feature='f-strings')
logger.debug('If you are using Python {}, prefer {feature} of course!', 3.6, feature='f-strings')

8、示例

from loguru import logger

logger.add('my_log.log') #加上该句会输出到文件，去掉则无
logger.info('普通信息')
logger.debug('调试消息')
logger.info('普通消息')
logger.warning('警告消息')
logger.error('错误消息')
logger.critical('严重错误')
logger.success('成功信息')

运行之后会发现目录下 my_log.log 出现了刚刚控制台输出的 DEBUG 信息，并且控制台也会显示。

 

二十、yagmail模块--自动发邮件

一般发邮件方法

import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 发送邮箱服务器
smtpserver = 'smtp.sina.com'
# 发送邮箱用户/密码
user = 'username@sina.com'
password = '123456'
# 发送邮箱
sender = 'username@sina.com'
# 接收邮箱
receiver = 'receive@126.com'
# 发送邮件主题
subject = 'Python email test'

# 编写HTML类型的邮件正文
msg = MIMEText('
你好！
','html','utf-8')
msg['Subject'] = Header(subject, 'utf-8')


# 连接发送邮件
smtp = smtplib.SMTP()
smtp.connect(smtpserver)
smtp.login(user, password)
smtp.sendmail(sender, receiver, msg.as_string())
smtp.quit()

其实，这段代码也并不复杂，只要你理解使用过邮箱发送邮件，那么以下问题是你必须要考虑的：

• 你登录的邮箱帐号/密码
• 对方的邮箱帐号
• 邮件内容（标题，正文，附件）
• 邮箱服务器（SMTP.xxx.com/pop3.xxx.com）

yagmail 可以更简单的来实现自动发邮件功能

安装pip install yagmail

简单例子

import yagmail

#链接邮箱服务器
yag = yagmail.SMTP( user="user@126.com", password="1234", host='smtp.126.com')

# 邮箱正文
contents = ['This is the body, and here is just text http://somedomain/image.png',
            'You can find an audio file attached.', '/local/path/song.mp3']

# 发送邮件
yag.send('taaa@126.com', 'subject', contents)

# 给多个用户发送邮件，只需要将接收邮箱 变成一个list即可。
yag.send(['aa@126.com','bb@qq.com','cc@gmail.com'], 'subject', contents)

发送带附件的邮件

import yagmail

#链接邮箱服务器
yag = yagmail.SMTP( user="user@126.com", password="1234", host='smtp.126.com')

# 邮箱正文
contents = ['This is the body, and here is just text http://somedomain/image.png',
            'You can find an audio file attached.', '/local/path/song.mp3']
# 发送邮件
yag.send('aaaa@126.com', '发送附件', contents, ["d://log.txt","d://baidu_img.jpg"])

只需要添加要发送的附件列表即可。

二十一、tenacity--错误自动重试

我们经常会遇到一些偶然发生的请求失败的状况，这种时候如果我们仅仅简单的捕捉错误然后跳过对应任务，肯定是不严谨的，尤其是在网络爬虫中，会存在损失有价值数据的风险。此时很有必要为我们的程序逻辑添加一些错误重试的策略，tenacity库可能是目前Python生态中最好用的错误重试库。https://github.com/jd/tenacity

1、tenacity的基础使用

tenacity的错误重试核心功能由其retry装饰器来实现，默认不给retry装饰器传参数时，它会在其所装饰的函数运行过程抛出错误时不停地重试下去

import random
from tenacity import retry

@retry
def demo_func1():
    a = random.random()
    print(a)
    if a >= 0.1:
        raise Exception

demo_func1()　

 当a小于0.1时程序结束，当a大于0.1时会抛出异常，然后会被tenacity不停的重试。

2 设置最大重试次数

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def demo_func2():
print('函数执行')
raise Exception

demo_func2()

3 设置最大重试超时时长

import time
from tenacity import retry, stop_after_delay

# 设置重试最大超时时长为5秒
@retry(stop=stop_after_delay(5))
def demo_func3():
    time.sleep(1)
    print(f'已过去 {time.time() - start_time} 秒')
    raise Exception

# 记录开始时间
start_time = time.time()
demo_func3()

4 组合重试停止条件
用|运算符组合不同的限制条件，实现最大重试次数以及最大超时时长限制，当函数执行重试超过3秒或次数大于5次时均可以结束重试：

import time
import random
from tenacity import retry, stop_after_delay, stop_after_attempt

@retry(stop=(stop_after_delay(3) | stop_after_attempt(5)))
def demo_func4():
    
    time.sleep(random.random())
    print(f'已过去 {time.time() - start_time} 秒')
    
    raise Exception

# 记录开始时间
start_time = time.time()
demo_func4()

5 设置相邻重试之间的时间间隔---固定时间

import time
from tenacity import retry, wait_fixed, stop_after_attempt

# 设置重试等待间隔为1秒
@retry(wait=wait_fixed(1), stop=stop_after_attempt(3))
def demo_func5():
    print(f'已过去 {time.time() - start_time} 秒')
    raise Exception
    
# 记录开始时间
start_time = time.time()
demo_func5()

6 设置相邻重试之间的时间间隔---随机时间

import time
from tenacity import retry, wait_random, stop_after_attempt

# 设置重试等待间隔为1到3之间的随机数
@retry(wait=wait_random(min=1, max=3), stop=stop_after_attempt(5))
def demo_func6():
    print(f'已过去 {time.time() - start_time} 秒')
    raise Exception

# 记录开始时间
start_time = time.time()
demo_func6()

7 捕捉或忽略特定的错误类型
使用tenacity中的retry_if_exception_type()和retry_if_not_exception_type()，配合retry()的retry参数，我们可以对特定的错误类型进行捕捉或忽略

from tenacity import retry, retry_if_exception_type, retry_if_not_exception_type

@retry(retry=retry_if_exception_type(FileExistsError))
def demo_func7():
    
    raise TimeoutError
    
@retry(retry=retry_if_not_exception_type(FileNotFoundError))
def demo_func8():

    raise FileNotFoundError

8 自定义函数结果条件判断函数
我们可以编写额外的条件判断函数，配合tenacity中的retry_if_result()，实现对函数的返回结果进行自定义条件判断，返回True时才会触发重试操作：

import random
from tenacity import retry, retry_if_result

@retry(retry=retry_if_result(lambda x: x >= 0.1))
def demo_func9():
    a = random.random()
    print(a)
    return a

# 记录开始时间
demo_func9()

二十二、schedule定时任务

import schedule
import time
 
def test1():
    print("I'm working...in test1")
    time.sleep(1)
    print("test1 end...")
    #return schedule.CancelJob         #加入该句则只会执行一次该任务

def test2():    
    print("I'm working...in test2")

#你能用day的地方一定能用days其他也一样，如果every()里面值不为1或者空时候必须用他复数
# schedule.every(5).seconds.do(test1)          # 每10秒执行一次job函数
# schedule.every(10).minutes.do(test1)         # 每10分钟执行一次job函数
# schedule.every().hours.do(test1)             #当every()没参数时默认是1小时/分/秒执行一次job函数
# schedule.every().days.at("20:04").do(test1)  # 每天的10:30执行一次job函数
# schedule.every(5).to(10).days.do(job)        #每隔5到10天(随机)执行一次任务
# schedule.every().minute.at(':17').do(test_job) #每分钟在第17秒的时候执行任务

# schedule.every(1).hours.until("18:30").do(job)              # 每个小时运行作业，18:30后停止
# schedule.every(1).hours.until("2030-01-01 18:33").do(job)   # 每个小时运行作业，2030-01-01 18:33 today
# schedule.every(1).hours.until(timedelta(hours=8)).do(job)   # 每个小时运行作业，8个小时后停止
# schedule.every(1).hours.until(time(11, 33, 42)).do(job)     # 每个小时运行作业，11:32:42后停止
# schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job) # 每个小时运行作业，2020-5-17 11:36:20后停止

# schedule.every().monday.do(test1)                # 每周一的这个时候执行一次job函数
# schedule.every().wednesday.at("13:15").do(test1) # 每个星期三的13:15执行一次job函数
# schedule.every().week.do(test1)                  # 每周凌晨运行一次

# schedule.every(10).seconds.do(test1,"参数") # 如果被执行函数有有参数时，这么写
# schedule.every(10).seconds.do(test1)          # 可以同时定时执行多个任务，但是每个任务是按顺序执行
# schedule.every(10).seconds.do(test2)          # 可以同时定时执行多个任务，但是每个任务是按顺序执行

# #取消指定任务,需要引入tap
# schedule.every().day.do(test1).tag('daily-tasks', 'friend')
# schedule.every().hour.do(test1).tag('hourly-tasks', 'friend')
# schedule.clear('daily-tasks') #q取消所有标签为daily-tasks的任务
# schedule.clear()   #取消任务

# schedule.run_all(delay_seconds=10)  # 立即运行所有作业，每次作业间隔10秒

while True:
    # 启动服务
    schedule.run_pending()
    time.sleep(1)

'''
I'm working...in test1
test1 end...
I'm working...in test2
I'm working...in test1
test1 end...
I'm working...in test2
I'm working...in test1
test1 end...
I'm working...in test2
'''

schedule方法是串行的，也就是说，如果各个任务之间时间不冲突，那是没问题的；如果时间有冲突的话，串行的执行会出问题，因此我们可以改成多线程的方式：

import schedule
import time
import threading

def job1():
    print("I'm working... in job1  start")
    time.sleep(15)
    print("job1 end...")

def job2():
    print("I'm working... in job2")
    print("job2 end...")

def run_threaded(job_func):
     job_thread = threading.Thread(target=job_func)
     job_thread.start()
       #threading.Thread(target=loop,args=(i,loops[i]))

schedule.every(10).seconds.do(run_threaded,job1)
schedule.every(10).seconds.do(run_threaded,job2)

while True:
    schedule.run_pending()
    time.sleep(1)

'''
I'm working... in job1  start
I'm working... in job2
job2 end...
I'm working... in job1  start
I'm working... in job2
job2 end...
job1 end...
I'm working... in job1  start
I'm working... in job2
job2 end...
'''

test_1.py

from fastapi import FastAPI
import uvicorn
  
app = FastAPI()
  
# 绑定路由和视图函数
#async def是定义异步函数的方法，你也可以定义为普通函数
@app.get("/")
async def index():
    return {"name": "Fastapi"}
  
@app.get("/int")
async def index1():
    return 666
  
@app.get("/str")
async def index2():
    return "wwwww"
  
@app.get("/bytes")
async def index3():
    return b"satori"
  
@app.get("/tuple")
async def index4():
    return ("EEEEE", "RRRRR", "TTTTT")
  
@app.get("/list")
async def index5():
    return [{"name": "DDDDD", "age": 17}, {"name": "HHHHH", "age": 16}]
  
if __name__ == "__main__":
# 启动服务，因为我们这个文件叫做main.py，所以需要启动main.py里面的app，然后是host和port表示监听的ip和端口reload、debug：表示可以自动重启
    uvicorn.run("test_1:app", host="0.0.0.0", port=5555,reload=True, debug=True)

　to_do.py

import schedule
import time,os
import webbrowser
from selenium import webdriver

def test1():
    os.system("start python d:\xxx\test_1.py")
    driver=webdriver.Chrome() #调用chrome浏览器
    driver.get('http://localhost:5555/docs')
    time.sleep(5)
    driver.quit()

schedule.every(10).seconds.do(test1)   
while True:
    # 启动服务
    schedule.run_pending()

二十三、APScheduler组件

APScheduler共有4种组件：

• 触发器（trigger），触发器中包含调度逻辑，每个作业都有自己的触发器来决定下次运行时间。除了它们自己初始配置以外，触发器完全是无状态的。
• 作业存储器（job store），存储被调度的作业，默认的作业存储器只是简单地把作业保存在内存中，其他的作业存储器则是将作业保存在数据库中，当作业被保存在一个持久化的作业存储器中的时候，该作业的数据会被序列化，并在加载时被反序列化，需要说明的是，作业存储器不能共享调度器。
• 执行器（executor），处理作业的运行，通常通过在作业中提交指定的可调用对象到一个线程或者进程池来进行，当作业完成时，执行器会将通知调度器。
• 调度器（scheduler），配置作业存储器和执行器可以在调度器中完成。例如添加、修改、移除作业，根据不同的应用场景，可以选择不同的调度器，可选的将在下一小节展示。

调度器

• BlockingScheduler : 当调度器是你应用中唯一要运行的东西时。
• BackgroundScheduler : 当你没有运行任何其他框架并希望调度器在你应用的后台执行时使用
• AsyncIOScheduler : 当你的程序使用了asyncio（一个异步框架）的时候使用。
• GeventScheduler : 当你的程序使用了gevent（高性能的Python并发框架）的时候使用。
• TornadoScheduler : 当你的程序基于Tornado（一个web框架）的时候使用。
• TwistedScheduler : 当你的程序使用了Twisted（一个异步框架）的时候使用
• QtScheduler : 如果你的应用是一个Qt应用的时候可以使用。

作业存储器

如果你的应用在每次启动的时候都会重新创建作业，那么使用默认的作业存储器（MemoryJobStore）即可，但是如果你需要在调度器重启或者应用程序奔溃的情况下任然保留作业，你应该根据你的应用环境来选择具体的作业存储器。例如：使用Mongo或者SQLAlchemy JobStore （用于支持大多数RDBMS）

执行器

对执行器的选择取决于你使用上面哪些框架，大多数情况下，使用默认的ThreadPoolExecutor已经能够满足需求。如果你的应用涉及到CPU密集型操作，你可以考虑使用ProcessPoolExecutor来使用更多的CPU核心。你也可以同时使用两者，将ProcessPoolExecutor作为第二执行器。

触发器

当你调度作业的时候，你需要为这个作业选择一个触发器，用来描述这个作业何时被触发，APScheduler有三种内置的触发器类型：

• date 一次性指定日期
• interval 在某个时间范围内间隔多长时间执行一次
• cron 和Linux crontab格式兼容，最为强大

interval 间隔调度（每隔多久执行）

weeks (int) – number of weeks to wait
days (int) – number of days to wait
hours (int) – number of hours to wait
minutes (int) – number of minutes to wait
seconds (int) – number of seconds to wait
start_date (datetime|str) – starting point for the interval calculation
end_date (datetime|str) – latest possible date/time to trigger on
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

#表示每隔3天17时19分07秒执行一次任务
sched.add_job(my_job, 'interval', days=03, hours=17, minutes=19, seconds=07)
#间隔3秒钟执行一次
scheduler.add_job(job3, 'interval', seconds=3)

2、date 定时调度（作业只会执行一次）

run_date (datetime|str) – 任务开始的时间）
timezone (datetime.tzinfo|str) – run_date的时区

scheduler.add_job(tick, 'date', run_date='2016-02-14 15:01:05')　　
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6))

 3、cron定时调度（某一定时时刻执行）

(int|str) 表示参数既可以是int类型，也可以是str类型
(datetime | str) 表示参数既可以是datetime类型，也可以是str类型
 
year (int|str) – 4-digit year -（表示四位数的年份，如2008年）
month (int|str) – month (1-12) -（表示取值范围为1-12月）
day (int|str) – day of the (1-31) -（表示取值范围为1-31日）
week (int|str) – ISO week (1-53) -（格里历2006年12月31日可以写成2006年-W52-7（扩展形式）或2006W527（紧凑形式））
day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) - （表示一周中的第几天，既可以用0-6表示也可以用其英语缩写表示）
hour (int|str) – hour (0-23) - （表示取值范围为0-23时）
minute (int|str) – minute (0-59) - （表示取值范围为0-59分）
second (int|str) – second (0-59) - （表示取值范围为0-59秒）
start_date (datetime|str) – earliest possible date/time to trigger on (inclusive) - （表示开始时间）
end_date (datetime|str) – latest possible date/time to trigger on (inclusive) - （表示结束时间）
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone) -（表示时区取值）

# 在 6,7,8,11,12 月的第三个周五 的 1:00,2:00,3:00 运行
scheduler.add_job(job1,'cron',month='6-8,11-12',day='3rd fri',hour='1-3')

# 在 2019-12-31 号之前的周一到周五 5 点 30 分运行
scheduler.add_job(job1,'cron',day_of_week='mon-fri',hour=5,minute=30,end_date='2019-12-31')

from apscheduler.schedulers.blocking import BlockingScheduler
import time


def job():
    print('开始运行……')
    time.sleep(3)
    print('运行结束啦……')


scheduler = BlockingScheduler(timezone='Asia/Shanghai')

# #每周的周一到周五的21:50执行一次任务
# scheduler.add_job(job, "cron", day_of_week="1-5", hour=21, minute=50)

# #每天的22:37 分执行一次任务
scheduler.add_job(job, 'cron', hour=22, minute=37)
# scheduler.add_job(job, 'cron', hour='22', minute='37')

# #表示 19:23、 20:23、 21:23 各执行一次任务
# scheduler.add_job(job, 'cron', day='4')

# #在1月,3月,5月,7-9月，每天的下午2点，每一分钟执行一次任务
# scheduler .add_job(job, 'cron', month='1,3,5,7-9', day='*', hour='14', minute='*')

# # 当前任务会在 6、7、8、11、12 月的第三个周五的 0、1、2、3 点执行
# scheduler .add_job(job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')

# #每300秒执行一次
# scheduler .add_job(job, 'interval', seconds=300)

# #从开始时间到结束时间，每隔俩小时运行一次
# scheduler .add_job(job, 'interval', hours=2, start_date='2018-01-10 09:30:00', end_date='2018-06-15 11:00:00')

scheduler.start()

删除 job
当调度器中删除 job 时，该 job 也将从其关联的 job 存储中删除，并且将不再执行。有两种方法可以实现此目的：

通过调用方法 remove_job() ，指定 job ID 和 job 存储别名
通过调用 add_job() 时 返回的 apscheduler.job.Job 实例的 remove() 方法
例：

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
或者：

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')
注意： 如果任务已经调度完毕，并且之后也不会再被执行的情况下，会被自动删除。

暂停和恢复 job
暂停和恢复 job 与 删除 job 方法类似：

暂停：

job = scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
job.pause() # or
scheduler.pause_job('my_job_id')
恢复：

job = scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
job.resume() # or
scheduler.resume_job('my_job_id')
获取 job 列表
使用 get_jobs() 方法获取一个列表，或者使用 print_jobs() 方法打印一个格式化的列表。

jobs = scheduler.get_jobs() # or
scheduler.print_jobs()
提示：可以使用 get_job(id) 获取单个 job 信息

from apscheduler.schedulers.blocking import BlockingScheduler
import time

def my_job(text=""):
    print(text, 'my_job1 is running')
    time.sleep(1)

sched = BlockingScheduler()

sched.add_job(my_job, 'interval', seconds=2, args=['第一个作业'])
sched.add_job(my_job, 'interval', seconds=2, id='my_job_id', args=['第二个作业'])

#获取job
# print(sched.get_jobs())
# print(sched.get_job(job_id="my_job_id"))
sched.print_jobs()

sched.start()

修改 job
修改 job 依然与 删除 job 方法类似，可以修改除 job id 以外的其他属性。

例：

job.modify(max_instances=6, name='Alternate name')
如果想修改触发器，可以使用 apscheduler.job.Job.reschedule 或者 apscheduler.schedulers.base.BaseScheduler.reschedule_job 。

例：

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')
其实修改 job 也可以使用 add_job() 方法，只需要指定参数 replace_existing=True 以及相同的 job_id 即可。

关闭调度器
关闭调度器方法：

scheduler.shutdown()
默认情况下，会关闭 job 存储和执行器，并等待所有正在执行的 job 完成。如果不想等待则可以使用以下方法关闭：

scheduler.shutdown(wait=False)
暂停/恢复调度器
暂停调度器：

scheduler.pause()
恢复调度器：

scheduler.resume()
启动调度器的时候可以指定 paused=True，以这种方式启动的调度器直接就是暂停状态。

scheduler.start(paused=True)