我们从Python开源项目中，提取了以下14个代码示例，用于说明如何使用asyncio.futures()。
def stream_from_fd(fd, loop): """Recieve a streamer for a given file descriptor.""" reader = asyncio.StreamReader(loop=loop) protocol = asyncio.StreamReaderProtocol(reader, loop=loop) waiter = asyncio.futures.Future(loop=loop) transport = UnixFileDescriptorTransport( loop=loop, fileno=fd, protocol=protocol, waiter=waiter, ) try: yield from waiter except: transport.close() raise if loop.get_debug(): logger.debug("Read fd %r connected: (%r, %r)", fd, transport, protocol) return reader, transport
def test_copy_state(self): from asyncio.futures import _copy_future_state f = asyncio.Future(loop=self.loop) f.set_result(10) newf = asyncio.Future(loop=self.loop) _copy_future_state(f, newf) self.assertTrue(newf.done()) self.assertEqual(newf.result(), 10) f_exception = asyncio.Future(loop=self.loop) f_exception.set_exception(RuntimeError()) newf_exception = asyncio.Future(loop=self.loop) _copy_future_state(f_exception, newf_exception) self.assertTrue(newf_exception.done()) self.assertRaises(RuntimeError, newf_exception.result) f_cancelled = asyncio.Future(loop=self.loop) f_cancelled.cancel() newf_cancelled = asyncio.Future(loop=self.loop) _copy_future_state(f_cancelled, newf_cancelled) self.assertTrue(newf_cancelled.cancelled())
def conn(self): self._check_engine() def coro_finished(f: asyncio.Future): if f.cancelled(): self.logger.warning('acquiring connection is cancelled') return e = f.exception() if e is not None and not isinstance(e, pymysql.err.OperationalError): self.logger.exception( 'Exception happened while acquiring connection: %s', str(e), exc_info=e) coro = asyncio.ensure_future(self.engine.acquire()) coro.add_done_callback(coro_finished) try: res = await asyncio.wait_for(asyncio.shield(coro), self.request_timeout, loop=self.loop) except asyncio.futures.TimeoutError as e: raise MySQLTimeoutError('Timeout error') from e return res
def begin(self, conn, *args, **kwargs): if conn is None: raise MySQLNotConnectedException('MySQL not connected') def coro_finished(f: asyncio.Future): if f.cancelled(): self.logger.warning('begin of trx is cancelled') return e = f.exception() if e is not None and not isinstance(e, pymysql.err.OperationalError): self.logger.error( 'Exception happened while beginning trx: %s', str(e), exc_info=e) coro = asyncio.ensure_future(conn.begin(*args, **kwargs), loop=self.loop) coro.add_done_callback(coro_finished) try: res = await coro return res except asyncio.futures.TimeoutError as e: raise MySQLTimeoutError('Timeout error') from e
def commit(self, trx, *args, **kwargs): if trx is None: raise ValueError('trx must be not None') def coro_finished(f: asyncio.Future): if f.cancelled(): self.logger.warning('commit of trx is cancelled') return e = f.exception() if e is not None and not isinstance(e, pymysql.err.OperationalError): self.logger.error( 'Exception happened while committing trx: %s', str(e), exc_info=e) coro = asyncio.ensure_future(trx.commit(*args, **kwargs), loop=self.loop) coro.add_done_callback(coro_finished) try: res = await coro except asyncio.futures.TimeoutError as e: raise MySQLTimeoutError('Timeout error') from e return res
def rollback(self, trx, *args, **kwargs): if trx is None: raise ValueError('trx must be not None') def coro_finished(f: asyncio.Future): if f.cancelled(): self.logger.warning('rollback of trx is cancelled') return e = f.exception() if e is not None and not isinstance(e, pymysql.err.OperationalError): self.logger.error( 'Exception happened while running rollback trx: %s', str(e), exc_info=e) coro = asyncio.ensure_future(trx.rollback(*args, **kwargs), loop=self.loop) coro.add_done_callback(coro_finished) try: res = await coro except asyncio.futures.TimeoutError as e: raise MySQLTimeoutError('Timeout error') from e return res
def test_in_use_flag(self): """ Do several blocking calls and see whether in_use increments. """ @asyncio.coroutine def test(): # Create connection connection = yield from self.create_pool(poolsize=10) for i in range(0, 10): yield from connection.delete(['my-list-%i' % i]) @asyncio.coroutine def sink(i): the_list, result = yield from connection.blpop(['my-list-%i' % i]) futures = [] for i in range(0, 10): self.assertEqual(connection.connections_in_use, i) futures.append(ensure_future(sink(i), loop=self.loop)) # Sleep to make sure that the above coroutine started executing. yield from asyncio.sleep(.1, loop=self.loop) # One more blocking call should fail. with self.assertRaises(NoAvailableConnectionsInPoolError) as e: yield from connection.delete(['my-list-one-more']) yield from connection.blpop(['my-list-one-more']) self.assertIn('No available connections in the pool', e.exception.args[0]) connection.close() # Consume this futures (which now contain ConnectionLostError) with self.assertRaises(ConnectionLostError): yield from asyncio.gather(*futures) self.loop.run_until_complete(test())
def __init__(self, args=None, loop=None, enable_logging=True): self._args = get_args(args) self._loop = loop self._enable_logging = enable_logging clean_dirs(self._args.buffer_dir) if self._args.clean_all: clean_dirs(self._args.logs_dir, self._args.cache_dir) setup_dirs(self._args.buffer_dir, self._args.logs_dir, self._args.cache_dir) # Create an event loop. self._autoclose_loop = False if self._loop is None: self._autoclose_loop = True self._loop = asyncio.new_event_loop() asyncio.set_event_loop(None) # Create custom executor. executor = concurrent.futures.ThreadPoolExecutor(self._args.max_workers) self._loop.set_default_executor(executor) if self._enable_logging: self._setup_logging(self._args.logs_dir, DEFAULT_SERVER_LOG_FILENAME, DEFAULT_ACCESS_LOG_FILENAME, debug=self._args.debug) self._cached_downloader = CachingDownloader(cache_dir=self._args.cache_dir, parallels=self._args.parallels, part_size=self._args.part_size, chunk_size=self._args.chunk_size, loop=self._loop)
def test_wrap_future(self): def run(arg): return (arg, threading.get_ident()) ex = concurrent.futures.ThreadPoolExecutor(1) f1 = ex.submit(run, 'oi') f2 = asyncio.wrap_future(f1, loop=self.loop) res, ident = self.loop.run_until_complete(f2) self.assertIsInstance(f2, asyncio.Future) self.assertEqual(res, 'oi') self.assertNotEqual(ident, threading.get_ident())
def test_wrap_future_use_global_loop(self, m_events): def run(arg): return (arg, threading.get_ident()) ex = concurrent.futures.ThreadPoolExecutor(1) f1 = ex.submit(run, 'oi') f2 = asyncio.wrap_future(f1) self.assertIs(m_events.get_event_loop.return_value, f2._loop)
def test_wrap_future_cancel(self): f1 = concurrent.futures.Future() f2 = asyncio.wrap_future(f1, loop=self.loop) f2.cancel() test_utils.run_briefly(self.loop) self.assertTrue(f1.cancelled()) self.assertTrue(f2.cancelled())
def test_wrap_future_cancel2(self): f1 = concurrent.futures.Future() f2 = asyncio.wrap_future(f1, loop=self.loop) f1.set_result(42) f2.cancel() test_utils.run_briefly(self.loop) self.assertFalse(f1.cancelled()) self.assertEqual(f1.result(), 42) self.assertTrue(f2.cancelled())
def execute(self, query, *multiparams, **params): """ ???????????? ??? ??????? ???????? conn ? ????? ????? """ conn = params.pop('conn') if conn is None: raise MySQLNotConnectedException('MySQL not connected') def coro_finished(f: asyncio.Future): if f.cancelled(): self.logger.warning( 'execute of query is cancelled. q: %s [%s, %s]', query, multiparams, params) return e = f.exception() if e is not None and not isinstance(e, pymysql.err.OperationalError): self.logger.error( 'Exception happened while ' 'executing query \'%s\': %s', query, str(e), exc_info=e) coro = asyncio.ensure_future( conn.execute(query, *multiparams, **params), loop=self.loop ) coro.add_done_callback(coro_finished) try: res = await coro return res except asyncio.futures.TimeoutError as e: raise MySQLTimeoutError('Timeout error') from e except pymysql.InternalError as e: err_code, err_msg = e.args if err_code == MySQLError.DEADLOCK: self.logger.error( 'Deadlock happened while executing query %s: %s', query, e.args) raise DeadlockError() from e raise e
def test_transaction(self, transport, protocol): # Prepare yield from protocol.set(u'my_key', u'a') yield from protocol.set(u'my_key2', u'b') yield from protocol.set(u'my_key3', u'c') yield from protocol.delete([u'my_hash']) yield from protocol.hmset(u'my_hash', {'a': '1', 'b': '2', 'c': '3'}) # Start transaction self.assertEqual(protocol.in_transaction, False) transaction = yield from protocol.multi() self.assertIsInstance(transaction, Transaction) self.assertEqual(protocol.in_transaction, True) # Run commands f1 = yield from transaction.get('my_key') f2 = yield from transaction.mget(['my_key', 'my_key2']) f3 = yield from transaction.get('my_key3') f4 = yield from transaction.mget(['my_key2', 'my_key3']) f5 = yield from transaction.hgetall('my_hash') for f in [f1, f2, f3, f4, f5]: self.assertIsInstance(f, Future) # Calling subscribe inside transaction should fail. with self.assertRaises(Error) as e: yield from transaction.start_subscribe() self.assertEqual(e.exception.args[0], 'Cannot start pubsub listener when a protocol is in use.') # Complete transaction result = yield from transaction.exec() self.assertEqual(result, None) self.assertEqual(protocol.in_transaction, False) # Read futures r1 = yield from f1 r3 = yield from f3 # 2 & 3 switched by purpose. (order shouldn't matter.) r2 = yield from f2 r4 = yield from f4 r5 = yield from f5 r2 = yield from r2.aslist() r4 = yield from r4.aslist() r5 = yield from r5.asdict() self.assertEqual(r1, u'a') self.assertEqual(r2, [u'a', u'b']) self.assertEqual(r3, u'c') self.assertEqual(r4, [u'b', u'c']) self.assertEqual(r5, {'a': '1', 'b': '2', 'c': '3'})
