Выборочная и динамическая типизация данных при выполнении INSERT с использованием python3 psycopg2 - PullRequest
Новая
       8

Выборочная и динамическая типизация данных при выполнении INSERT с использованием python3 psycopg2

0 голосов
/ 07 ноября 2018

Мне нужно синхронизировать любую таблицу postgres из последней копии нашей производственной базы данных (источника) в базу данных разработчика, не стирая ее данные тестирования. Приведенный ниже упрощенный код работает для большинства таблиц, но не для таблиц с полями jsonb из-за psycopg2.ProgrammingError: невозможно адаптировать тип 'dict' 

    import psycopg2
    from psycopg2 import sql

    tb = "table_to_be_copied"

    ##############################
    # load data from source DB                                                                                                                          
    ##############################

    conn_source = psycopg2.connect(host='source_localhost',
                             dbname=postgres,
                             user='xyz',
                             port=source_port)

    cursor_source = conn_source.cursor()
    cursor_source.execute(
        sql.SQL("SELECT * from {}").format(sql.Identifier(tb))
    )

    # obtain column names on the fly for any given table
    colnames = tuple([desc[0] for desc in cursor_source.description])        

    # jsonb's type code is 3802. This will help the program determine on the fly
    # which columns are in jsonb.
    typecodes = tuple([desc[1] for desc in cursor_source.description])

    # obtain production data to be synced
    rows = cursor_source.fetchall()

    cursor_source.close()
    conn_source.close()

    ##############################
    # upsert data into destination DB
    ##############################

    conn_dest = psycopg2.connect(host='dest_localhost',
                                 dbname='postgres',
                                 user='xyz',
                                 port=dest_port)

    cursor_dest = conn_dest.cursor()

    for row in rows:
            cursor_dest.execute(
                    sql.SQL("INSERT INTO {} ({}) VALUES ({}) \                                                                                          
                    ON CONFLICT (id) DO UPDATE SET ({}) = ({})").format(
                            sql.Identifier(tb),
                            sql.SQL(', ').join(map(sql.Identifier, colnames)),
                            sql.SQL(', ').join(sql.Placeholder() * len(colnames)),
                            sql.SQL(', ').join(map(sql.Identifier, colnames)),
                            sql.SQL(', ').join(sql.Placeholder() * len(colnames))),
                    row * 2)

    conn_dest.commit()
    cursor_dest.close()
    conn_dest.close()

    print ("Sync done")

Если это возможно, я предпочитаю не извиняться за 2 запроса: один UPSERT для полей не-jsonb и должен обрабатывать поля jsonb NOT NULL; другое ОБНОВЛЕНИЕ для JSONB Typecasting.

Спасибо.

...