dbapi_hook target_fields not working

[galzak]

I have a table in Postgres with the below columns:
re_ads_snapshot_id serial NOT NULL,
ad_id int8 NOT NULL,
origin_table_id int4 NULL

My ETL suppose to insert data into this table but only to ad_id, origin_table_id.
the query is “select ad_id, origin_table_id from….”
When I use target fields it’s still inserting the Data into the first two: re_ads_snapshot_id and ad_id instead of skipping the first serial column.

Does anyone know how can i skip the first column? since it's serial number..

That is my mySQL to Postgres operator code:

class MySQLToPostgresOperator(BaseOperator):
    @apply_defaults
    def __init__(
            self,
            sql,
            pg_table,
            src_mysql_conn_id,
            dest_postgres_conn_id,
            target_fields=None,
            *args, **kwargs):
        super(MySQLToPostgresOperator, self).__init__(*args, **kwargs)
        self.sql = sql
        self.pg_table = pg_table
        self.src_mysql_conn_id = src_mysql_conn_id
        self.dest_postgres_conn_id = dest_postgres_conn_id
        self.target_fields = target_fields
    def execute(self, context):
        log.info("Executing: {}".format(str(self.sql)))
        src_mysql = MySqlHook(mysql_conn_id=self.src_mysql_conn_id)
        dest_pg = PostgresHook(postgres_conn_id=self.dest_postgres_conn_id)
        log.info("Transferring MySQL query results into Postgres database.")
        conn = src_mysql.get_conn()
        cursor = conn.cursor()
        cursor.execute(self.sql)
        log.info("Inserting rows into Postgres")
        dest_pg.insert_rows(table=self.pg_table, rows=cursor, target_fields=self.target_fields)
        log.info("Done.")

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[eladkal]

Commented about it on Slack.
Your implementation seems to be a private case of GenericTransfer so you should be able to use that operator for your ETL.

I think for now this issue can be closed.

[galzak]

Thank you.
instead of using:

        conn = src_mysql.get_conn()
        cursor = conn.cursor()
        cursor.execute(self.sql)

I had to change:
records = src_mysql.get_records(self.sql)