Olá!! Estou com esse problema, alguém sabe me ajudar? Apesar de ter colocado a API key no Aiflow, não está aparecendo quando rodo o twitter_hook. Os outros campos vem normal. Também está acontecendo outros erros de conversão para o Json pelo que entendi...

Meu código está um pouco diferente na parte de construir url mas acredito que não seja esse o problema:

from airflow.hooks.http_hook import HttpHook
import requests
import json

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

    def create_url(self):
        base_url = self.base_url
        query = self.query
        start_time = {
            self.start_time
            if self.start_time
            else ""
        }
        end_time = {
            self.end_time
            if self.end_time
            else ""
        }
        query_params = [f'query={query}',
        'tweet.fields=author_id,conversation_id,created_at,id,in_reply_to_user_id,public_metrics,text',
        'expansions=author_id',
        'user.fields=id,name,username,created_at',
        f'start_time={start_time}&end_time={end_time}']
        search_url = base_url+f'2/tweets/search/recent?{query_params[0]}&{query_params[1]}&{query_params[2]}&{query_params[3]}&{query_params[4]}'

        return search_url

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

    def paginate(self, search_url,session,next_token=""):
        if next_token:
            full_url = f"{search_url}&next_token={next_token}"
        else:
            full_url = search_url
        data = self.connect_to_endpoint(full_url,session) #json
        yield data 
        if "next_token" in data.get("meta",{}):
            yield from self.paginate(search_url,session,data['meta']['next_token'])

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

        yield from self.paginate(url,session)


if __name__ == "__main__":
    for pg in TwitterHook("AluraOnline").run():
        print(json.dumps(pg,ident=4,sort_keys=True))

Restante do erro: