Extra[10] - Agendando tarefas com Python
comentada em 19/07/2020 19:52

Resposta do @Orionte hoje aqui na área me inspirou a fazer este tópico aqui sobre agendamento de tarefas com Python. 

Eu já tinha pensado em fazer o tópico antes mas a lib padrão que vem na linguagem usa conceitos de threads e process e achei que seria interessante deixar pra depois, mas hoje pesquisando achei outra lib externa chamada "schedule" que tem sintaxe super simples então serve pra gente brincar aqui hoje.

Imaginem que criamos um programa que faz web scraping de determinada página na internet. Ou até aqui na Bastter, por exemplo, um script que roda a cada tempo procurando dados novos na B3 de balanços das empresas. 

Imaginem que esse meu programa captura dados e eu quero sempre ter esses dados salvos em relatórios feitos em planilhas. 

Então para isso, vou precisar das libs xlsxwriter e schedule:

$ python -m venv venv
$ source venv/bin/activate
$ pip install xlsxwriter schedule
$ pip freeze
schedule==0.6.0
XlsxWriter==1.1.8

Iniciando nosso programa, vou criar uma planilha qualquer mas, como serão criadas várias planilhas diferentes dentro da mesma pasta, preciso diferenciá-las pelo nome. Para organizar os nomes, vou adicionar ao nome da planilha dados de data e hora com a lib datetime:

import xlsxwriter
from datetime import datetime
import schedule
# datetime.today().isoformat() retorna data e hora no seguinte formato
# '2019-08-05T21:46:17.361509'
current_date_and_time = datetime.today().isoformat()

# Abaixo é realizado o desempacotamento de iteráveis com atribuição de valores às variáveis
(year, month, day, hour, minute, second) = ( 
                                            current_date_and_time[0:4], 
                                            current_date_and_time[5:7],  
                                            current_date_and_time[8:10], 
                                            current_date_and_time[11:13], 
                                            current_date_and_time[14:16], 
                                            current_date_and_time[17:19] 
                                            ) 
filename = 'excel_file'
sheet = 'excel_sheet'

# Cria o arquivo
workbook = xlsxwriter.Workbook(f'{filename}_{hour}{minute}{second}_{day}{month}{year}.xlsx')
# Cria a planilha dentro do arquivo
worksheet = workbook.add_worksheet(f'{sheet}')

Agora imaginem que meus dados estejam dentro de uma lista hipotética:

from random import randint

# Código até o momento

data = [randint(1, 100) for x in range(100)]  # Cria uma lista com 100 números aleatórios de 1 a 100

Terminando o código de escrever em nossa planilha:

# Código até o momento

worksheet.write(0, 0, 'Data de coleta')  # Escreve na linha 0 e coluna 0 a string ao lado
worksheet.write(1, 0, f'{hour}:{minute}:{second} - {day}/{month}/{year}')  # Linha 1, coluna 0, string ao lado
worksheet.write(0, 1, 'Dados')
worksheet.write_column('B2', data)  # Escreve a lista a partir da célula B2

workbook.close()  # Fecha a planilha

Se rodarmos o código até agora, ele simplesmente salva uma planilha qualquer com a lista gerada. Mas queremos automatizar essa tarefa, então vou precisar da lib schedule. No site da lib tem alguns exemplos de uso dela com uma legibilidade impressionante:

import schedule
import time

def job():
    print("I'm working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)

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

Nem preciso comentar muito sobre o código, a legibilidade é tão linda que é auto explicável. 

O que cada linha iniciada em "schedule.every..." diz é que a cada intervalo de tempo dado o scheduler adiciona no schedule a função a ser realizada. Dentro do laço logo ao final, a função "schedule.run_pending()" roda as tarefas que estão na fila do scheduler. Se nenhuma tarefa estiver agendada, nada é feito.

Basta adicionar nosso código até o momento numa estrutura similar a essa:

import schedule
import time
import xlsxwriter
from datetime import datetime
from random import randint
def creates_worksheet():

    # datetime.today().isoformat() retorna data e hora no seguinte formato
    # '2019-08-05T21:46:17.361509'
    current_date_and_time = datetime.today().isoformat()
    (year, month, day, hour, minute, second) = ( 
                                                current_date_and_time[0:4], 
                                                current_date_and_time[5:7],  
                                                current_date_and_time[8:10], 
                                                current_date_and_time[11:13], 
                                                current_date_and_time[14:16], 
                                                current_date_and_time[17:19] 
                                                ) 

    filename = 'excel_file'
    sheet = 'excel_sheet'

    # Cria o arquivo
    workbook = xlsxwriter.Workbook(f'{filename}_{hour}{minute}{second}_{day}{month}{year}.xlsx')
    # Cria a planilha dentro do arquivo
    worksheet = workbook.add_worksheet(f'{sheet}')

    data = [randint(1, 100) for x in range(100)]
    worksheet.write(0, 0, 'Data de coleta')
    worksheet.write(1, 0, f'{hour}:{minute}:{second} - {day}/{month}/{year}')
    worksheet.write(0, 1, 'Dados')
    worksheet.write_column('B2', data)

    workbook.close()

schedule.every(0.05).minutes.do(creates_worksheet)  # A cada 3 segundos adiciona o agendamento da execução da função acima dentro do scheduler

while True:
    schedule.run_pending()  # Roda as tarefas agendadas
    time.sleep(1)  # Espera um segundo pra rodar o laço novamente

Um gif da pasta do arquivo com o script em execução: 



À primeira vista a gente pode pensar que nem é necessário usar uma lib externa para fazer esse trabalho. Basta dar um time.sleep() no laço que o resultado é parecido, mas o interessante de se usar libs de agendamento de tarefas é que elas podem agendar diversas tarefas de uma vez e incluir na pilha de tarefas definindo prioridades. Fazer esse tipo de tarefa com um sleep dentro de um laço não é nada interessante.