Run query and insert the result to another table
psycopg2 already implements your extract_table

Since you're using cursor.copy_from to load data into the new table, why not use something along the lines of cursor.copy_to to write them to a file?

However, cursor.copy_to will translate to COPY table_name TO 'filename' [ [ WITH ] ( option [, ...] ) ] in its current implementation. You thus need to use cursor.copy_expert and provide the whole SQL query to be executed.

The copy_expert call will write to STDOUT or read from STDIN and redirect to/from the file-like object passed as the second argument (file). I believe the query to execute should look like

copy_query = "COPY (" + query + ")"\
    "TO STDOUT"\
    "WITH CSV HEADER DELIMITER '\t' NULL '' ENCODING 'utf-8'"


with or without the HEADER option; not sure how copy_from would handle it.

You can thus rewrite copy_table something along the lines of:

# Using format string because we trust the query, are we?
copy_query = "COPY ({}) TO STDOUT WITH CSV HEADER DELIMITER '{}' NULL '' ENCODING '{}'".format(src_query, DELIMITER, ENCODING)
with NamedTemporaryFile('w+t', encoding=ENCODING, newline='') as fp:
    logging.info("Writing result to a temporary file.")
    with src_conn.cursor() as cursor:
        cursor.copy_expert(copy_query, fp)
    fp.flush()
    fp.seek(0)
    logging.info("Copying result to {}".format(dst_table_name))
    with dst_conn.cursor() as cursor:
        if truncate_table:
            logging.info("Truncating {}.".format(dst_table_name))
            # It seems like we do trust the new name too
            cursor.execute('TRUNCATE {};'.format(dst_table_name))
        cursor.copy_from(file=fp, table=dst_table_name, sep=DELIMITER, null="")
    dst_conn.commit()


Handling database state

The dst_conn.commit() at the end of copy_table is bad practice. You most likely want to use the context manager available on connection objects to handle that for you, especially if things goes wrong.

Using the connection's context manager is to handle commits and rollbacks only and does not close the connection when leaving the context block. So you can write, say

conn = psycopg2.connect(...)

with conn:
    # Use cursor(s)

with conn:
    # Use (other) cursor(s)

with conn:
    # You get the deal

conn.close()


You're also handling the creation of named cursors using lot of duplicated code. I understand that their use differ a bit from client cursors, but the creation can be simplified to:

cursor = conn.cursor(name=get_random_cursor_name() if use_named_cursor else None)


since the name argument is already None by default.

You can even go further by letting the user provide its own name using constructs such as:

def copy_table(..., cursor_name=None, ...):
    ...
    with src_conn:
        with src_conn.cursor(name=cursor_name) as cursor:
            ...


This way, you can have calls such as:


copy_table(src_conn, dst_conn, query, dst_table_name='foobar')
copy_table(src_conn, dst_conn, query, cursor_name='foobar', dst_table_name='foobar')
copy_table(src_conn, dst_conn, query, cursor_name=get_random_cursor_name(17), dst_table_name='foobar')


Protect from user input

As usual in SQL, you should not trust user input. As such, you might want to not directly use the provided query string. However you do want to build the copy query yourself. For these usages, psycopg2 provides the cursor.mogrify method.

The copy query can thus be built using:

copy_query = "COPY (%s) TO STDOUT WITH CSV HEADER DELIMITER '%s' NULL '' ENCODING '%s'"


and then passed to copy_expert with:

cursor.copy_expert(cursor.mogrify(copy_query, (src_query, DELIMITER, ENCODING)), fp)


You may also be interested in providing a way to build the source query by binding parameters. In such cases, variable length arguments can be used to simplify handling of bound arguments:

def copy_table(query, *args, src_conn, dest_conn, dest_table_name,
               cursor_name=None, truncate_table=False):
    copy_query = "COPY (%s) TO STDOUT WITH CSV HEADER DELIMITER '%s' NULL '' ENCODING '%s'"
    with NamedTemporaryFile('w+t', encoding=ENCODING, newline='') as fp:
        logging.info("Writing result to a temporary file.")
        with src_conn:
            with src_conn.cursor(name=cursor_name) as cursor:
                query = cursor.mogrify(query, args)
                copy_query = cursor.mogrify(copy_query, (query, DELIMITER, ENCODING))
                cursor.copy_expert(copy_query, fp)
        fp.flush()
        fp.seek(0)
        logging.info("Copying result to {}".format(dst_table_name))
        with dest_conn:
            with dest_conn.cursor() as cursor:
                if truncate_table:
                    logging.info("Truncating {}.".format(dst_table_name))
                    cursor.execute('TRUNCATE %s;', (dst_table_name,))
                cursor.copy_from(file=fp, table=dst_table_name, sep=DELIMITER, null="")


You will need to use named arguments for anything other than the query and its parameters:

copy_table('SELECT * FROM public.my_table WHERE product_count &gt; %s', 200,
           src_conn=database_remote,
           dest_conn=database_local,
           dest_table_name='public.my_table_subset',
           cursor_name='foobar')