[Bug] strftime não esta formatando data_interval_start do jinja | Apache Airflow: extração de dados

Boa noite , eu estou com um erro no airflow que no momento que construo minha url pelo DAG , como fiz no vídeo eu estou usando os parametros de start_time e end_time com as variáveis do jinja , mas qnd faço a formatação do padrão do twiter : (start_time="{{ data_interval_start.strftime('%Y-%m-%dT%H:%M:%S.00Z') }}", ) , minha url esta ficando destorcida, ja fiz o teste rodando no python direto e funciona normal sem o arquivo do DAG , mas qnd uso o DAG não esta formatando a data com a função strftime, teve alguma mudança , tem algum ajuste pra ser feito ?

[2024-06-19, 03:53:20 UTC] {local_task_job_runner.py:120} ▶ Pre task execution logs
[2024-06-19, 03:53:21 UTC] {base.py:84} INFO - Using connection ID 'twitter_default' for task execution.
[2024-06-19, 03:53:21 UTC] {twitter_hook.py:35} INFO - URL : https://labdados.com/2/tweets/search/recent?query=datascience&tweet.fields=author_id,conversation_id,creted_at,id,in_replay_to_user_id,public_metrics,lang,text&expansions=author_id&user.fields=id,name,username,creted_at&start_time={{ data_interval_start.strftime('%Y-%m-%dT%H:%M:%S.00Z') }}&end_time={{ data_interval_end.strftime('%Y-%m-%dT%H:%M:%S.00Z') }}
[2024-06-19, 03:53:22 UTC] {http.py:199} ERROR - HTTP error: Bad Gateway
[2024-06-19, 03:53:22 UTC] {http.py:200} ERROR - <html>
<head><title>502 Bad Gateway</title></head>
<body>
<center><h1>502 Bad Gateway</h1></center>
<hr><center>nginx/1.18.0 (Ubuntu)</center>
</body>
</html>

Oii, Enrico! Tudo bem?

Poxa, sinto muito pelo impedimento que está tento. Não se preocupe, vamos buscar algumas alternativas.

Em um primeiro momento, recomendo que compare o seu código com o código da aula, para verificar se possui algum erro de sintaxe ou lógica que possa estar interferindo nesta comunicação. Segue o trecho do código:

import sys
sys.path.append("airflow_pipeline")

from airflow.models import DAG
from datetime import datetime, timedelta
from operators.twitter_operator import TwitterOperator
from os.path import join   
from airflow.utils.dates import days_ago
        
with DAG(dag_id = "TwitterDAG", start_date=days_ago(2), schedule_interval="@daily") as dag:

    TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.00Z"
    query = "datascience"

    to = TwitterOperator(file_path=join("datalake/twitter_datascience",
        "extract_data={{ ds }}",
        "datascience_{{ ds_nodash }}.json"),
        query=query, 
    start_time="{{ data_interval_start.strftime('%Y-%m-%dT%H:%M:%S.00Z') }}", 
    end_time="{{ data_interval_end.strftime('%Y-%m-%dT%H:%M:%S.00Z') }}", 
    task_id="twitter_datascience")

Outra possível solução, é garantir que o Jinja esteja interpretando corretamente essas variáveis. Você pode tentar utilizar a função format do Jinja diretamente na string, em vez de usar strftime diretamente. Veja como você pode fazer isso:

start_time="{{ data_interval_start | ds_format('%Y-%m-%dT%H:%M:%S.00Z') }}"
end_time="{{ data_interval_end | ds_format('%Y-%m-%dT%H:%M:%S.00Z') }}"

No exemplo acima o ds_format é um filtro do Jinja que formata a data conforme o padrão especificado.

Outra abordagem seria definir a formatação da data dentro do próprio código Python antes de passar para o template do Jinja. Por exemplo:

from airflow.utils.dates import days_ago
from datetime import datetime, timedelta

def format_datetime(dt):
    return dt.strftime('%Y-%m-%dT%H:%M:%S.00Z')

with DAG(dag_id="TwitterDAG", start_date=days_ago(2), schedule_interval="@daily") as dag:
    TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.00Z"
    query = "datascience"

    start_time = "{{ data_interval_start | ds_format('%Y-%m-%dT%H:%M:%S.00Z') }}"
    end_time = "{{ data_interval_end | ds_format('%Y-%m-%dT%H:%M:%S.00Z') }}"

    to = TwitterOperator(
        file_path=join("datalake/twitter_datascience", "extract_data={{ ds }}", "datascience_{{ ds_nodash }}.json"),
        query=query,
        start_time=start_time,
        end_time=end_time,
        task_id="twitter_datascience"
    )

Não esqueça de garantir que o ds_format esteja disponível no contexto do seu template. Caso contrário, você pode criar um filtro personalizado no Jinja para realizar essa formatação.

Espero que umas das sugestões te ajude, faça os testes e observe se o projeto funciona como esperado.

Bons estudos, Enrico!