Как выполнять параллельные запросы с PyGre SQL?

Question

Я пытаюсь выполнить несколько запросов параллельно с PyGre SQL и многопроцессорностью, но приведенный ниже код зависает без возврата:

from pg import DB
from multiprocessing import Pool
from functools import partial


def create_query(table_name):
  return f"""create table {table_name} (id integer);
  CREATE INDEX ON {table_name} USING BTREE (id);"""

my_queries = [ create_query('foo'), create_query('bar'), create_query('baz') ]


def execute_query(conn_string, query):
  con = DB(conn_string)
  con.query(query)
  con.close()

rs_conn_string = "host=localhost port=5432 dbname=postgres user=postgres password="
pool = Pool(processes=len(my_queries))
pool.map(partial(execute_query,rs_conn_string), my_queries)

Есть ли способ заставить его работать? Также возможно ли сделать 3 запущенных запроса в одной «транзакции» в случае сбоя одного запроса и отката другого?

Answer 1

Одна очевидная проблема заключается в том, что вы всегда запускаете pool.map не только в основном процессе, но и тогда, когда интерпретаторы, используемые в параллельных подпроцессах, импортируют сценарий. Вместо этого вы должны сделать что-то вроде этого:

def run_all():
    with Pool(processes=len(my_queries)) as pool:
        pool.map(partial(execute_query,rs_conn_string), my_queries)

if __name__ == '__main__':
    run_all()

Что касается вашего второго вопроса, это невозможно, поскольку транзакция выполняется для каждого соединения, которые находятся в отдельных процессах, если вы делаете это так.

Возможно, вам нужна асинхронная обработка команд, но еще не поддерживается PyGre SQL. Psygopg + aiopg , вероятно, лучше подходит для подобных вещей.