erro ao testar o script twitter_hook_py

o código é:

from airflow.providers.http.hooks.http import HttpHook
import requests
from datetime import datetime, timedelta
import json


class TwitterHook(HttpHook):
    def __init__(self, end_time, start_time, query, conn_id=None):
        self.end_time = end_time
        self.start_time = start_time
        self.query = query
        self.conn_id = conn_id or "twitter_default"
        super().__init__(http_conn_id=self.conn_id)


    def create_url(self):
        TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.00Z"

        end_time   = self.end_time
        start_time = self.start_time
        query      = self.query

        tweet_fields = "tweet.fields=author_id,conversation_id,created_at,id,in_reply_to_user_id,public_metrics,lang,text"
        user_fields = "expansions=author_id&user.fields=id,name,username,created_at"

        url_raw = f"{self.base_url}/2/tweets/search/recent?query={query}&{tweet_fields}&{user_fields}&start_time={start_time}&end_time={end_time}" 

        return url_raw  


    def connect_to_endpoint(self, url, session):
        request = requests.Request("GET", url)
        prep = session.prepare_request(request)
        self.log.info(f"URL: {url}")

        return self.run_and_check(session, prep, {})


    def paginate(self, url_raw, session):
        lista_json_response = []

        #Imprimindo json
        response = self.connect_to_endpoint(url_raw, session)

        json_response = response.json()

        lista_json_response.append(json_response)

        #Paginate
        contador = 1

        while "next_token" in json_response.get("meta",{}) and contador<100:
            next_token = json_response['meta']['next_token']
            url = f"{url_raw}&next_token={next_token}"
            response = self.connect_to_endpoint(url, session) 
            json_response = response.json()
            lista_json_response.append(json_response)

            contador += 1 

        return lista_json_response


    def run(self):
        session = self.get_conn()
        url_raw = self.create_url()


        return self.paginate(url_raw, session)



if __name__ == "__main__":

   #montando url
   TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.00Z"

   end_time = datetime.now().strftime(TIMESTAMP_FORMAT)
   start_time = (datetime.now() +      timedelta(-1)).date().strftime(TIMESTAMP_FORMAT)   
   query = "datascience"
   for pg in TwitterHook(end_time, start_time, query).run():
       print(json.dumps(pg,indent=4, sort_keys=True))

Olá Rogério, tudo bem ? Espero que sim.

Desculpa pela demora no retorno.

Pela mensagem de erro, está tudo certo com o Hook e com a requisição, o problema está na autentificação com a API do Twiiter.

2022-12-10 20:07:00,516] {http.py:142} ERROR - HTTP error: Unauthorized
[2022-12-10 20:07:00,536] {http.py:143} ERROR - {
  "title": "Unauthorized",
  "type": "about:blank",
  "status": 401,
  "detail": "Unauthorized"
}

Sugiro que verifique se o Bearer Token está registrado na conexão criada no Airflow.

Para isso basta navegamos na aba Admin, em seguida vai aparecer um botão chamado Connections e então podemos editar a conexão twitter_default.

Na tela de edição vamos verificar o campo Extra.

Nós trabalhamos nessa conexão na Aula 3 atividade 7.

Um problema comum que pode acontecer é o navegador preencher as informações dos campos Login e Password, verifique se esses campos estão em branco antes de salvar.

Espero ter ajudado, mas se o problema persistir pode mandar mensagem aqui no fórum que vamos juntos resolver.

Bons estudos.