Python - AttributeError: объект 'NoneType' не имеет атрибута 'execute' - PullRequest
Новая
       10

Python - AttributeError: объект 'NoneType' не имеет атрибута 'execute'

0 голосов
/ 30 мая 2018

Я пытаюсь запустить скрипт на python, который входит в Amazon Redshift DB и затем выполняет команду SQL.Я использую инструмент под названием Airflow для управления рабочим процессом.При выполнении приведенного ниже кода я могу нормально войти в БД, но при попытке выполнить команду SQL выдается следующая ошибка: 

**AttributeError: 'NoneType' object has no attribute 'execute'**

Код: 

## Login to DB

def db_log(**kwargs):
  global db_con
  try:
    db_con = psycopg2.connect(
       " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = '5439'")
  except:
    print("I am unable to connect")
    print('Connection Task Complete')
    task_instance = kwargs['task_instance']
    task_instance.xcom_push(key="dwh_connection" , value = "dwh_connection")
    return (dwh_connection)


def insert_data(**kwargs):
  task_instance = kwargs['task_instance']
  db_con_xcom = task_instance.xcom_pull(key="dwh_connection", task_ids='DWH_Connect')
  cur = db_con_xcom
  cur.execute("""insert into tbl_1 select limit 2 ;""")

Может ли кто-нибудь помочь мне исправить это.Спасибо ..

Полный код: 

## Third party Library Imports
import pandas as pd
import psycopg2
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from sqlalchemy import create_engine
import io

# Following are defaults which can be overridden later on
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2018, 5, 29, 12),
'email': ['airflow@airflow.com']
}

dag = DAG('sample1', default_args=default_args)

## Login to DB

def db_log(**kwargs):
  global db_con
  try:
    db_con = psycopg2.connect(
       " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = '5439'")
  except:
    print("I am unable to connect")
    print('Connection Task Complete')
    task_instance = kwargs['task_instance']
    task_instance.xcom_push(key="dwh_connection" , value = "dwh_connection")
    return (dwh_connection)



t1 = PythonOperator(
  task_id='DWH_Connect',
  python_callable=data_warehouse_login,provide_context=True,
  dag=dag)

#######################

def insert_data(**kwargs):
  task_instance = kwargs['task_instance']
  db_con_xcom = task_instance.xcom_pull(key="dwh_connection", task_ids='DWH_Connect')
  cur = db_con_xcom
  cur.execute("""insert into tbl_1 select limit 2 """)


##########################################

t2 = PythonOperator(
  task_id='DWH_Connect1',
  python_callable=insert_data,provide_context=True,dag=dag)

t1 >> t2

1 Ответ

0 голосов
/ 30 мая 2018

Вы уверены, что добавили весь свой код?Вы вызываете функцию data_warehouse_login в python_callable первой задачи, но она не определена.Предполагая, что это должно быть db_log, и первая задача была успешной, вы на самом деле ничего не добавляете ко второй задаче (поскольку ваш xcom_push срабатывает только при ошибке).

Обычно не будетв любом случае не советую делать xcom-объект подключения.В качестве альтернативы вы можете рассмотреть возможность использования прилагаемого PostgresHook, который должен охватывать ваш вариант использования и одинаково хорошо работает с Amazon Redshift.

https://github.com/apache/incubator-airflow/blob/master/airflow/hooks/postgres_hook.py

...