Airflow PostgresHook возвращает идентификатор из оператора вставки, не фиксирующего

Question

Я использую PostgresHook в операторе Airflow.

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)

Это возвращает Id, но запись не зафиксирована в таблице действий.Я попытался запустить get_records и get_first, и ни один из них не зафиксировал.

.run фиксирует, но не возвращает идентификатор результата.

Это правильный способ вставить запись и затем вернуть идентификатор?

Answer 1

Вы можете позвонить get_autocommit(), чтобы проверить, включена ли автоматическая фиксация, а затем set_autocommit(), чтобы включить явно. Похоже, что Airflow DBApiHook наивно полагает, что вы не будете ничего фиксировать при извлечении записей. Установка этого значения должна решить эту проблему.

Если вы хотите еще больше контролировать происходящее, вы можете вызвать get_conn() или get_cursor() и повторить логику, которая происходит внутри run() и get_first(), для фиксации вручную.