[Bug] Erro de conexão?

Segue meu 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}https://labdados.com/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 = []
        #imprimir json
        response = self.connect_to_endpoint(url_raw, session)
        json_response = response.json()
        lista_json_response.append (json_response)
        contador = 1

        #paginate
        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(days=-7)).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))

Duas coisas que notei:

1. ao abrir o Airflow no navegador, a senha que eu havia salvo pra logar como admin mudou. Precisei pegar a nova senha ao rodar o "airflow standalone".
2. a conexão que eu já havia criado "twitter_default" sumiu, mas mesmo criando-a novamente o mesmo erro mencionado aqui no post persiste.

O erro mencionado anteriormente a princípio foi superado.

Na atividade 12 da aula 3 vocês colocaram o código export AIRFLOW_HOME=$(pwd)/airflow_pipeline pra exportar a variável e, por isso, eu estava subindo outra instância do Airflow, sendo que o código correto pro meu caso é a seguinte export AIRFLOW_HOME=~/Documents/airflowalura.

Agora um novo erro está sendo apresentado, segue:

(venv) poletto@poletto-virtual-machine:~/Documents/CURSO1$ python3.9 airflow_pipeline/hook/twitter_hook.py
Traceback (most recent call last):
  File "/home/poletto/Documents/CURSO1/airflow_pipeline/hook/twitter_hook.py", line 70, in <module>
    for pg in TwitterHook (end_time, start_time, query).run():
  File "/home/poletto/Documents/CURSO1/airflow_pipeline/hook/twitter_hook.py", line 57, in run
    session = self.get_conn()
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/providers/http/hooks/http.py", line 100, in get_conn
    conn = self.get_connection(self.http_conn_id)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/hooks/base.py", line 72, in get_connection
    conn = Connection.get_connection_from_secrets(conn_id)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/models/connection.py", line 434, in get_connection_from_secrets
    raise AirflowNotFoundException(f"The conn_id `{conn_id}` isn't defined")
airflow.exceptions.AirflowNotFoundException: The conn_id `twitter_default` isn't defined

1 semana da primeira postagem e nenhuma resposta ainda

Estou com o mesmo problema do amigo: airflow.exceptions.AirflowNotFoundException: The conn_id twitter_default isn't defined

Alguém poderia ajudar?

Estou com o mesmo problema do colega:

Olá Marcelo, Marcos e Rafael.

Peço desculpas pela demora na resposta.

Analisando as mensagens aqui notei dois pontos.

• O primeiro, no código passado na segunda mensagem do Marcelo, temos a utilização do self.base_url junto com a URL completa o que vai gerar uma URL invalida já que ele vai combinar a raiz da URL duas vezes (https://labdados.com). A variável self.base_url já armazena esse valor, não é necessário repeti-lo na url_raw.

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

Então removendo a raiz da URL ficamos com esse código.

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

• O segundo ponto, quanto ao erro airflow.exceptions.AirflowNotFoundException: The conn_id twitter_default isn't defined, ele normalmente aparece quando não definimos a variável AIRFLOW_HOME.

O comando de referencia é o export combinado com o pwd, mas é muito importante você ter certeza que no terminal você está na pasta que contem a pasta do seus dados airflow, que pode ser airflowalura ou airflow_pipeline. (pode rodar o comando dir para verificar se uma dessas pastas aparece)

export AIRFLOW_HOME=$(pwd)/airflowalura

Outra maneira que você pode ter acesso ao caminho da pasta do AIRFLOW é clicando com o botão direito em cima dela e copiando seu caminho.

E depois colar esse caminho na criação da variável. (Esse caminho é do projeto na minha maquina, na de vocês será diferente.)

export AIRFLOW_HOME=/home/alura/Documentos/curso-1/Material/Aula3/airflow_pipeline

Depois de criar a variável você pode testar seu Hook.

python3.9 airflow_pipeline/hook/twitter_hook_aluno.py 

Caso o erro persista, nesse mesmo terminal você pode rodar o comando airflow standalone e confirmar se sua conexão foi criada corretamente.

Caso não encontre a conexão crie ela novamente conforme Aula 3 - Atividade 7 e depois de criar a conexão, no mesmo terminal, utilizar as teclas Ctrl + C para encerrar o Airflow e rode novamente o comando para testar seu Hook.

Acredito que esses passos devem resolver o erro.

Fico no aguardo da confirmação, mas caso o erro persista, compartilhem o erro completo aqui no fórum e também o código do Hook e vamos juntos buscar outras alternativas.

Peço novamente desculpa pela demora na resposta.

Bom dia, Igor! Obrigado pelo retorno!

Fiz as correções que tu mencionou, inclusive precisei criar novamente a conexão twitter_default, mas o erro persiste. Segue ele completo em duas partes pelo limite de carcteres:

(venv) poletto@poletto-virtual-machine:~/Documents/CURSO1/airflow_pipeline$ python3.9 hook/twitter_hook.py 
[2023-06-12T09:56:15.482-0300] {connection.py:428} ERROR - Unable to retrieve connection from secrets backend (MetastoreBackend). Checking subsequent secrets backend.
Traceback (most recent call last):
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context
    self.dialect.do_execute(
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
sqlite3.OperationalError: no such table: connection

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/models/connection.py", line 424, in get_connection_from_secrets
    conn = secrets_backend.get_connection(conn_id=conn_id)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/secrets/metastore.py", line 41, in get_connection
    conn = session.query(Connection).filter(Connection.conn_id == conn_id).first()
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/orm/query.py", line 2824, in first
    return self.limit(1)._iter().first()
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/orm/query.py", line 2916, in _iter
    result = self.session.execute(
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 1717, in execute
    result = conn._execute_20(statement, params or {}, execution_options)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1710, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1577, in _execute_clauseelement
    ret = self._execute_context(
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1948, in _execute_context
    self._handle_dbapi_exception(
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 2129, in _handle_dbapi_exception
    util.raise_(
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context
    self.dialect.do_execute(
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: connection
[SQL: SELECT connection.password AS connection_password, connection.extra AS connection_extra, connection.id AS connection_id, connection.conn_id AS connection_conn_id, connection.conn_type AS connection_conn_type, connection.description AS connection_description, connection.host AS connection_host, connection.schema AS connection_schema, connection.login AS connection_login, connection.port AS connection_port, connection.is_encrypted AS connection_is_encrypted, connection.is_extra_encrypted AS connection_is_extra_encrypted 

FROM connection 
WHERE connection.conn_id = ?
 LIMIT ? OFFSET ?]
[parameters: ('twitter_default', 1, 0)]
(Background on this error at: https://sqlalche.me/e/14/e3q8)
Traceback (most recent call last):
  File "/home/poletto/Documents/CURSO1/airflow_pipeline/hook/twitter_hook.py", line 70, in <module>
    for pg in TwitterHook (end_time, start_time, query).run():
  File "/home/poletto/Documents/CURSO1/airflow_pipeline/hook/twitter_hook.py", line 57, in run
    session = self.get_conn()
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/providers/http/hooks/http.py", line 100, in get_conn
    conn = self.get_connection(self.http_conn_id)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/hooks/base.py", line 72, in get_connection
    conn = Connection.get_connection_from_secrets(conn_id)
  File "/home/poletto/Documents/CURSO1/venv/lib/python3.9/site-packages/airflow/models/connection.py", line 434, in get_connection_from_secrets
    raise AirflowNotFoundException(f"The conn_id `{conn_id}` isn't defined")
airflow.exceptions.AirflowNotFoundException: The conn_id `twitter_default` isn't defined

Segue o código e o caminho completo do meu arquivo /home/poletto/Documents/CURSO1/airflow_pipeline/twitter_hook.py

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 = []
        #imprimir json
        response = self.connect_to_endpoint(url_raw, session)
        json_response = response.json()
        lista_json_response.append (json_response)
        contador = 1

        #paginate
        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(days=-7)).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))

Comandos que eu utilizei no terminal:

poletto@poletto-virtual-machine:~/Documents$ dir
airflowalura  CURSO1  datapipeline
poletto@poletto-virtual-machine:~/Documents$ cd CURSO1/
poletto@poletto-virtual-machine:~/Documents/CURSO1$ source venv/bin/activate
(venv) poletto@poletto-virtual-machine:~/Documents/CURSO1$ cd airflow_pipeline/
(venv) poletto@poletto-virtual-machine:~/Documents/CURSO1/airflow_pipeline$ export AIRFLOW_HOME=$(pwd)/airflow_pipeline
(venv) poletto@poletto-virtual-machine:~/Documents/CURSO1/airflow_pipeline$ python3.9 hook/twitter_hook.py 

Como fiquei na dúvida quanto a essa parte da tua mensagem:

Caso não encontre a conexão crie ela novamente conforme Aula 3 - Atividade 7 e depois de criar a conexão, no mesmo terminal, utilizar as teclas Ctrl + C para encerrar o Airflow e rode novamente o comando para testar seu Hook.

acabei executando o comando (venv) poletto@poletto-virtual-machine:~/Documents/CURSO1/airflow_pipeline$ python3.9 hook/twitter_hook.py antes e depois de encerrar o Airflow e o erro aparentemente é o mesmo.

Olá Marcelo.

Acredito que o erro está acontecendo porque estão sendo criados mais de um ambiente Airflow.

Repetindo os comandos que você enviou em um novo ambiente tenho essa situação. Uma pasta chamada aiflow_pipeline que contem apenas a pasta Hook.

Quando no terminal eu uso o comando export AIRFLOW_HOME=$(pwd)/airflow_pipeline eu estou dizendo que o nosso airflow deve rodar de uma pasta chamada airflow_pipeline que ficara dentro da pasta atual, que no nosso caso tambem é a airflow_pipeline.

Então quando rodarmos nosso script ou o comando airflow standalone vamos ter como resultado a criação de uma nova pasta chamada aiflow_pipeline, que não é o resultado que pretendíamos. (E que no caso também não tem a conexão twitter_default criada)

Para corrigir isso vamos primeiro apagar a pasta airflow_pipeline que criamos, clicando com o botão direito do mouse e selecionar a opção excluir permanentemente (cuidado para não clicar na pasta airflow_pipeline original que é onde está nossa pasta Hook.

Para corrigir o caminho do airflow, podemos rodar novamente o export e tirar o nome da pasta (mas isso só funciona porque já estamos dentro da pasta airflow_pipeline)

export AIRFLOW_HOME=$(pwd)/

Rodando novamente nosso script veremos que alguns novos arquivos vão ser criados dentro da nossa pasta, indicando que o airflow está rodando nessa pasta, porem ainda teremos o erro de conexão, porque ela ainda não existe nesse Airflow.

Devemos iniciar novamente o Airflow e configurar a conexão twitter_default. Novos arquivos serão criados nas pasta, como por exemplo o arquivo de senha do airflow.

airflow standalone

Depois de configurar o Airflow nosso terminal fica "travado" nessa tarefa, para encerrar o Airflow usamos as teclas Ctrl + C no terminal. Dessa maneira vamos poder executar novos comandos nesse terminal, que já está com o ambiente ativo e já tem a variável AIRFLOW_HOME exportada corretamente.

Por fim podemos rodar nosso script novamente, que agora vai encontrar a pasta correta do airflow, que é a que já configuramos a conexão AIRFLOW_HOME.

python3.9 hook/twitter_hook.py

Sei que as questões de variável de ambientes e as pastas do Airflow pode ser algo confuso no inicio, então fica tranquilo caso não de certo de primeira, tente novamente sempre atento as pastas criadas e sua localização no terminal para entender como o Airflow está funcionando. Se não conseguir descobrir pode me mandar um print do terminal e das pastas que vamos analisar juntos o que está acontecendo.

Bom dia, Igor!

Passando para informar que agora deu certo!

Realmente estava fazendo confusão com os locais onde rodar os comandos e isso estava fazendo com que novas pastas fossem criadas.

Seguindo as orientações do teu último post deu certinho!

Obrigado pela atenção e paciência.