python asyncio future_Python asyncio.isfuture方法代码示例
本文整理匯總了Python中asyncio.isfuture方法的典型用法代碼示例。如果您正苦于以下問題：Python asyncio.isfuture方法的具體用法？Python asyncio.isfuture怎么用？Python asyncio.isfuture使用的例子？那么恭喜您, 這里精選的方法代碼示例或許可以為您提供幫助。您也可以進(jìn)一步了解該方法所在模塊asyncio的用法示例。
在下文中一共展示了asyncio.isfuture方法的10個(gè)代碼示例，這些例子默認(rèn)根據(jù)受歡迎程度排序。您可以為喜歡或者感覺有用的代碼點(diǎn)贊，您的評(píng)價(jià)將有助于我們的系統(tǒng)推薦出更棒的Python代碼示例。
示例1: test_isfuture
?點(diǎn)贊 6
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def test_isfuture(self):
class MyFuture:
_asyncio_future_blocking = None
def __init__(self):
self._asyncio_future_blocking = False
self.assertFalse(asyncio.isfuture(MyFuture))
self.assertTrue(asyncio.isfuture(MyFuture()))
self.assertFalse(asyncio.isfuture(1))
self.assertFalse(asyncio.isfuture(asyncio.Future))
# As `isinstance(Mock(), Future)` returns `False`
self.assertFalse(asyncio.isfuture(mock.Mock()))
# As `isinstance(Mock(Future), Future)` returns `True`
self.assertTrue(asyncio.isfuture(mock.Mock(asyncio.Future)))
f = asyncio.Future(loop=self.loop)
self.assertTrue(asyncio.isfuture(f))
f.cancel()
開發(fā)者ID:ShikyoKira，項(xiàng)目名稱:Project-New-Reign---Nemesis-Main，代碼行數(shù):24，
示例2: test_isfuture
?點(diǎn)贊 6
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def test_isfuture(self):
class MyFuture:
_asyncio_future_blocking = None
def __init__(self):
self._asyncio_future_blocking = False
self.assertFalse(asyncio.isfuture(MyFuture))
self.assertTrue(asyncio.isfuture(MyFuture()))
self.assertFalse(asyncio.isfuture(1))
# As `isinstance(Mock(), Future)` returns `False`
self.assertFalse(asyncio.isfuture(mock.Mock()))
f = self._new_future(loop=self.loop)
self.assertTrue(asyncio.isfuture(f))
self.assertFalse(asyncio.isfuture(type(f)))
# As `isinstance(Mock(Future), Future)` returns `True`
self.assertTrue(asyncio.isfuture(mock.Mock(type(f))))
f.cancel()
開發(fā)者ID:bkerler，項(xiàng)目名稱:android_universal，代碼行數(shù):24，
示例3: test_api_calls_return_a_response_when_run_in_sync_mode
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def test_api_calls_return_a_response_when_run_in_sync_mode(self):
self.client.token = "xoxb-api_test"
resp = self.client.api_test()
self.assertFalse(asyncio.isfuture(resp))
self.assertTrue(resp["ok"])
開發(fā)者ID:slackapi，項(xiàng)目名稱:python-slackclient，代碼行數(shù):7，
示例4: test_api_calls_return_a_future_when_run_in_async_mode
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def test_api_calls_return_a_future_when_run_in_async_mode(self):
self.client.token = "xoxb-api_test"
self.client.run_async = True
future = self.client.api_test()
self.assertTrue(asyncio.isfuture(future))
resp = await future
self.assertEqual(200, resp.status_code)
self.assertTrue(resp["ok"])
開發(fā)者ID:slackapi，項(xiàng)目名稱:python-slackclient，代碼行數(shù):10，
示例5: _dispatch
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def _dispatch(self, f, value=None):
self._check_exhausted()
if f is None:
return
elif asyncio.isfuture(f):
f.set_result(value)
elif asyncio.iscoroutinefunction(f):
self.loop.create_task(f(value))
else:
f(value)
# self.loop.call_soon(functools.partial(f, value))
開發(fā)者ID:zh217，項(xiàng)目名稱:aiochan，代碼行數(shù):14，
示例6: __init__
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def __init__(self, identifier, hashable_key, type_spec, target_future):
"""Creates a cached value.
Args:
identifier: An instance of `CachedValueIdentifier`.
hashable_key: A hashable source value key, if any, or `None` of not
applicable in this context, for use during cleanup.
type_spec: The type signature of the target, an instance of `tff.Type`.
target_future: An asyncio future that returns an instance of
`executor_value_base.ExecutorValue` that represents a value embedded in
the target executor.
Raises:
TypeError: If the arguments are of the wrong types.
"""
py_typecheck.check_type(identifier, CachedValueIdentifier)
py_typecheck.check_type(hashable_key, collections.Hashable)
py_typecheck.check_type(type_spec, computation_types.Type)
if not asyncio.isfuture(target_future):
raise TypeError('Expected an asyncio future, got {}'.format(
py_typecheck.type_string(type(target_future))))
self._identifier = identifier
self._hashable_key = hashable_key
self._type_spec = type_spec
self._target_future = target_future
self._computed_result = None
開發(fā)者ID:tensorflow，項(xiàng)目名稱:federated，代碼行數(shù):28，
示例7: test_mock_from_create_future
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def test_mock_from_create_future(self, klass):
loop = asyncio.new_event_loop()
try:
if not (hasattr(loop, "create_future") and
hasattr(asyncio, "isfuture")):
return
mock = klass(loop.create_future())
self.assertTrue(asyncio.isfuture(mock))
finally:
loop.close()
開發(fā)者ID:Martiusweb，項(xiàng)目名稱:asynctest，代碼行數(shù):14，
示例8: isfuture
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def isfuture(fut):
return isinstance(fut, asyncio.Future)
開發(fā)者ID:skylander86，項(xiàng)目名稱:lambda-text-extractor，代碼行數(shù):4，
示例9: test_start_stop
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
def test_start_stop(self):
self.assertTrue(asyncio.isfuture(self.order_book_tracker._order_book_snapshot_router_task))
self.order_book_tracker.stop()
self.assertIsNone(self.order_book_tracker._order_book_snapshot_router_task)
self.order_book_tracker.start()
開發(fā)者ID:CoinAlpha，項(xiàng)目名稱:hummingbot，代碼行數(shù):7，
示例10: test_wrap_future
?點(diǎn)贊 5
?
# 需要導(dǎo)入模塊: import asyncio [as 別名]
# 或者: from asyncio import isfuture [as 別名]
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.assertTrue(asyncio.isfuture(f2))
self.assertEqual(res, 'oi')
self.assertNotEqual(ident, threading.get_ident())
ex.shutdown(wait=True)
開發(fā)者ID:bkerler，項(xiàng)目名稱:android_universal，代碼行數(shù):14，
注：本文中的asyncio.isfuture方法示例整理自Github/MSDocs等源碼及文檔管理平臺(tái)，相關(guān)代碼片段篩選自各路編程大神貢獻(xiàn)的開源項(xiàng)目，源碼版權(quán)歸原作者所有，傳播和使用請(qǐng)參考對(duì)應(yīng)項(xiàng)目的License；未經(jīng)允許，請(qǐng)勿轉(zhuǎn)載。
總結(jié)
以上是生活随笔為你收集整理的python asyncio future_Python asyncio.isfuture方法代码示例的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： ddos攻击测试网站（云ddos攻击在线
- 下一篇： python计算方位角_实例讲解：用py