"""A Future class similar to the one in PEP 3148."""__all__ = ('CancelledError', 'TimeoutError', 'InvalidStateError','Future', 'wrap_future', 'isfuture',)import concurrent.futuresimport contextvarsimport loggingimport sysfrom . import base_futuresfrom . import eventsfrom . import format_helpersCancelledError = base_futures.CancelledErrorInvalidStateError = base_futures.InvalidStateErrorTimeoutError = base_futures.TimeoutErrorisfuture = base_futures.isfuture_PENDING = base_futures._PENDING_CANCELLED = base_futures._CANCELLED_FINISHED = base_futures._FINISHEDSTACK_DEBUG = logging.DEBUG - 1 # heavy-duty debuggingclass Future:"""This class is *almost* compatible with concurrent.futures.Future.Differences:- This class is not thread-safe.- result() and exception() do not take a timeout argument andraise an exception when the future isn't done yet.- Callbacks registered with add_done_callback() are always calledvia the event loop's call_soon().- This class is not compatible with the wait() and as_completed()methods in the concurrent.futures package.(In Python 3.4 or later we may be able to unify the implementations.)"""# Class variables serving as defaults for instance variables._state = _PENDING_result = None_exception = None_loop = None_source_traceback = None# This field is used for a dual purpose:# - Its presence is a marker to declare that a class implements# the Future protocol (i.e. is intended to be duck-type compatible).# The value must also be not-None, to enable a subclass to declare# that it is not compatible by setting this to None.# - It is set by __iter__() below so that Task._step() can tell# the difference between# `await Future()` or`yield from Future()` (correct) vs.# `yield Future()` (incorrect)._asyncio_future_blocking = False__log_traceback = Falsedef __init__(self, *, loop=None):"""Initialize the future.The optional event_loop argument allows explicitly setting the eventloop object used by the future. If it's not provided, the future usesthe default event loop."""if loop is None:self._loop = events.get_event_loop()else:self._loop = loopself._callbacks = []if self._loop.get_debug():self._source_traceback = format_helpers.extract_stack(sys._getframe(1))_repr_info = base_futures._future_repr_infodef __repr__(self):return '<{} {}>'.format(self.__class__.__name__,' '.join(self._repr_info()))def __del__(self):if not self.__log_traceback:# set_exception() was not called, or result() or exception()# has consumed the exceptionreturnexc = self._exceptioncontext = {'message':f'{self.__class__.__name__} exception was never retrieved','exception': exc,'future': self,}if self._source_traceback:context['source_traceback'] = self._source_tracebackself._loop.call_exception_handler(context)@propertydef _log_traceback(self):return self.__log_traceback@_log_traceback.setterdef _log_traceback(self, val):if bool(val):raise ValueError('_log_traceback can only be set to False')self.__log_traceback = Falsedef get_loop(self):"""Return the event loop the Future is bound to."""return self._loopdef cancel(self):"""Cancel the future and schedule callbacks.If the future is already done or cancelled, return False. Otherwise,change the future's state to cancelled, schedule the callbacks andreturn True."""self.__log_traceback = Falseif self._state != _PENDING:return Falseself._state = _CANCELLEDself.__schedule_callbacks()return Truedef __schedule_callbacks(self):"""Internal: Ask the event loop to call all callbacks.The callbacks are scheduled to be called as soon as possible. Alsoclears the callback list."""callbacks = self._callbacks[:]if not callbacks:returnself._callbacks[:] = []for callback, ctx in callbacks:self._loop.call_soon(callback, self, context=ctx)def cancelled(self):"""Return True if the future was cancelled."""return self._state == _CANCELLED# Don't implement running(); see http://bugs.python.org/issue18699def done(self):"""Return True if the future is done.Done means either that a result / exception are available, or that thefuture was cancelled."""return self._state != _PENDINGdef result(self):"""Return the result this future represents.If the future has been cancelled, raises CancelledError. If thefuture's result isn't yet available, raises InvalidStateError. Ifthe future is done and has an exception set, this exception is raised."""if self._state == _CANCELLED:raise CancelledErrorif self._state != _FINISHED:raise InvalidStateError('Result is not ready.')self.__log_traceback = Falseif self._exception is not None:raise self._exceptionreturn self._resultdef exception(self):"""Return the exception that was set on this future.The exception (or None if no exception was set) is returned only ifthe future is done. If the future has been cancelled, raisesCancelledError. If the future isn't done yet, raisesInvalidStateError."""if self._state == _CANCELLED:raise CancelledErrorif self._state != _FINISHED:raise InvalidStateError('Exception is not set.')self.__log_traceback = Falsereturn self._exceptiondef add_done_callback(self, fn, *, context=None):"""Add a callback to be run when the future becomes done.The callback is called with a single argument - the future object. Ifthe future is already done when this is called, the callback isscheduled with call_soon."""if self._state != _PENDING:self._loop.call_soon(fn, self, context=context)else:if context is None:context = contextvars.copy_context()self._callbacks.append((fn, context))# New method not in PEP 3148.def remove_done_callback(self, fn):"""Remove all instances of a callback from the "call when done" list.Returns the number of callbacks removed."""filtered_callbacks = [(f, ctx)for (f, ctx) in self._callbacksif f != fn]removed_count = len(self._callbacks) - len(filtered_callbacks)if removed_count:self._callbacks[:] = filtered_callbacksreturn removed_count# So-called internal methods (note: no set_running_or_notify_cancel()).def set_result(self, result):"""Mark the future done and set its result.If the future is already done when this method is called, raisesInvalidStateError."""if self._state != _PENDING:raise InvalidStateError('{}: {!r}'.format(self._state, self))self._result = resultself._state = _FINISHEDself.__schedule_callbacks()def set_exception(self, exception):"""Mark the future done and set an exception.If the future is already done when this method is called, raisesInvalidStateError."""if self._state != _PENDING:raise InvalidStateError('{}: {!r}'.format(self._state, self))if isinstance(exception, type):exception = exception()if type(exception) is StopIteration:raise TypeError("StopIteration interacts badly with generators ""and cannot be raised into a Future")self._exception = exceptionself._state = _FINISHEDself.__schedule_callbacks()self.__log_traceback = Truedef __await__(self):if not self.done():self._asyncio_future_blocking = Trueyield self # This tells Task to wait for completion.if not self.done():raise RuntimeError("await wasn't used with future")return self.result() # May raise too.__iter__ = __await__ # make compatible with 'yield from'.# Needed for testing purposes._PyFuture = Futuredef _get_loop(fut):# Tries to call Future.get_loop() if it's available.# Otherwise fallbacks to using the old '_loop' property.try:get_loop = fut.get_loopexcept AttributeError:passelse:return get_loop()return fut._loopdef _set_result_unless_cancelled(fut, result):"""Helper setting the result only if the future was not cancelled."""if fut.cancelled():returnfut.set_result(result)def _set_concurrent_future_state(concurrent, source):"""Copy state from a future to a concurrent.futures.Future."""assert source.done()if source.cancelled():concurrent.cancel()if not concurrent.set_running_or_notify_cancel():returnexception = source.exception()if exception is not None:concurrent.set_exception(exception)else:result = source.result()concurrent.set_result(result)def _copy_future_state(source, dest):"""Internal helper to copy state from another Future.The other Future may be a concurrent.futures.Future."""assert source.done()if dest.cancelled():returnassert not dest.done()if source.cancelled():dest.cancel()else:exception = source.exception()if exception is not None:dest.set_exception(exception)else:result = source.result()dest.set_result(result)def _chain_future(source, destination):"""Chain two futures so that when one completes, so does the other.The result (or exception) of source will be copied to destination.If destination is cancelled, source gets cancelled too.Compatible with both asyncio.Future and concurrent.futures.Future."""if not isfuture(source) and not isinstance(source,concurrent.futures.Future):raise TypeError('A future is required for source argument')if not isfuture(destination) and not isinstance(destination,concurrent.futures.Future):raise TypeError('A future is required for destination argument')source_loop = _get_loop(source) if isfuture(source) else Nonedest_loop = _get_loop(destination) if isfuture(destination) else Nonedef _set_state(future, other):if isfuture(future):_copy_future_state(other, future)else:_set_concurrent_future_state(future, other)def _call_check_cancel(destination):if destination.cancelled():if source_loop is None or source_loop is dest_loop:source.cancel()else:source_loop.call_soon_threadsafe(source.cancel)def _call_set_state(source):if (destination.cancelled() anddest_loop is not None and dest_loop.is_closed()):returnif dest_loop is None or dest_loop is source_loop:_set_state(destination, source)else:dest_loop.call_soon_threadsafe(_set_state, destination, source)destination.add_done_callback(_call_check_cancel)source.add_done_callback(_call_set_state)def wrap_future(future, *, loop=None):"""Wrap concurrent.futures.Future object."""if isfuture(future):return futureassert isinstance(future, concurrent.futures.Future), \f'concurrent.futures.Future is expected, got {future!r}'if loop is None:loop = events.get_event_loop()new_future = loop.create_future()_chain_future(future, new_future)return new_futuretry:import _asyncioexcept ImportError:passelse:# _CFuture is needed for tests.Future = _CFuture = _asyncio.Future
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。