3

I am using PostgresHook in an Airflow operator.

pg_hook = PostgresHook(postgres_conn_id='postgres_default')
insert_activities_sql = "INSERT INTO activities (---) VALUES (---) RETURNING id "


activity_results = pg_hook.get_first(insert_activities_sql,parameters=insert_activities_params)

This does return the Id but the record is not committed into the activities table. I have tried running get_records and get_first and neither commit.

.run commits but does not return the results id.

Is this the correct way to insert a record and then return the id?

Murray Bryant
  • 45
  • 1
  • 3

1 Answers1

3

You can call get_autocommit() to check whether or not autocommit is enabled and then set_autocommit() to enable explicitly. It appears that the Airflow DBApiHook is naively assuming you will not be committing anything when fetching records. Setting it explicitly should resolve that issue.

If you would like even more control over what is happening, you can call get_conn() or get_cursor() and replicate the logic that is happening inside of run() and get_first() to manually commit.

andscoop
  • 939
  • 5
  • 8
  • Thanks for the pointer. pg_hook.get_autocommit() returned an error. But as you suggested I just replicated the logic inside the run and manually committed – Murray Bryant Nov 15 '18 at 01:58
  • @MurrayBryant is your code in a place where you would be able to share it (or at least a generic version of it)? I've been struggling through the same commit issue with get_pandas_df() and am having trouble coding this without executing the same SQL twice. Similarly, pg_hook.get_autocommit() didn't work for me. Thanks! – Dan Dec 11 '18 at 21:24