log = logging.getLogger()
def make_connection(options, asynchronous=False):
- return psycopg2.connect(dbname=options.dbname, user=options.user,
- password=options.password, host=options.host,
- port=options.port, async_=asynchronous)
+ params = {'dbname' : options.dbname,
+ 'user' : options.user,
+ 'password' : options.password,
+ 'host' : options.host,
+ 'port' : options.port,
+ 'async' : asynchronous}
+
+ return psycopg2.connect(**params)
class RankRunner(object):
def wait(self):
""" Block until any pending operation is done.
"""
- wait_select(self.conn)
- self.current_query = None
+ while True:
+ try:
+ wait_select(self.conn)
+ self.current_query = None
+ return
+ except psycopg2.extensions.TransactionRollbackError as e:
+ if e.pgcode == '40P01':
+ log.info("Deadlock detected (params = {}), retry."
+ .format(self.current_params))
+ self.cursor.execute(self.current_query, self.current_params)
+ else:
+ raise
+ except psycopg2.errors.DeadlockDetected:
+ self.cursor.execute(self.current_query, self.current_params)
def perform(self, sql, args=None):
""" Send SQL query to the server. Returns immediately without
self.cursor.execute(self.current_query, self.current_params)
else:
raise
+ except psycopg2.errors.DeadlockDetected:
+ self.cursor.execute(self.current_query, self.current_params)
return False