O código permite a criação das pastas com as referidas datas de extração, mas os arquivos como os dados climáticos não são gerados. Segue código utilizado para análise.
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
import pendulum
from os.path import join
import pandas as pd
from airflow.macros import ds_add
with DAG(
"dados_climaticos",
start_date = pendulum.datetime(2025, 2, 24, tz="UTC"),
schedule_interval='0 0 * * 1',# Executa toda segunda feira. Foi utilizado a cron expression sendo que (minutos = 0, horas = 0, dia do mês = *, mês = *, dia da semana = 1 que representa a segunda feira) )
) as dag:
tarefa_1 = BashOperator(
task_id = 'cria_pasta',
bash_command = 'mkdir -p "/home/frederico/Documents/airflow/pasta={{data_interval_end.strftime("%y-%m-%d")}}"'
)
def extrai_dados(data_interval_end):
city = 'Boston'
key = 'JU68JYL7FHEUT7JH8AF3YL4P7'
URL = join('https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/',
f'{city}/{data_interval_end}/{ds_add(data_interval_end,7)}?unitGroup=metric&include=days&key={key}&contentType=csv')
dados = pd.read_csv(URL)
file_path = f'/home/frederico/Documents/airflow/semana={data_interval_end}/'
# Salvando os dados
dados.to_csv(file_path + 'dados_brutos.csv')
dados[['datetime','tempmin','temp','tempmax']].to_csv(file_path + 'temperatura.csv')
dados[['datetime','description', 'icon']].to_csv(file_path + 'condicoes.csv')
tarefa_2 = PythonOperator(
task_id = 'extrai_dados',
python_callable = extrai_dados,
op_kwargs = {'data_interval_end':'{{data_interval_end.strftime("%Y-%m-%d")}}'}
)
tarefa_1 >> tarefa_2
Revisei o código conforme a transcrição da aula mas não indentifiquei a falha do mesmo. Também fiz um teste para verificar se o limite de requições já havvia sido atingido, mas não foi atingido.