- fetcher = DBConnection(self.dsn, cursor_factory=psycopg2.extras.DictCursor)
- psycopg2.extras.register_hstore(fetcher.conn,
- oid=hstore_oid,
- array_oid=hstore_array_oid)
-
- with WorkerPool(self.dsn, self.num_threads) as pool:
- places = self._fetch_next_batch(cur, fetcher, runner)
- while places is not None:
- if not places:
- t0 = time.time()
- fetcher.wait()
- fetcher_wait += time.time() - t0
- places = fetcher.cursor.fetchall()
+ with PlaceFetcher(self.dsn, conn) as fetcher:
+ with WorkerPool(self.dsn, self.num_threads) as pool:
+ has_more = fetcher.fetch_next_batch(cur, runner)
+ while has_more:
+ places = fetcher.get_batch()