Python typing 模块，Awaitable() 实例源码

def waitFor(self, selectorOrFunctionOrTimeout: Union[str, int, float],
                options: dict = None, **kwargs: Any) -> Awaitable:
        """Wait until `selectorOrFunctionOrTimeout`."""
        if options is None:
            options = dict()
        options.update(kwargs)
        if isinstance(selectorOrFunctionOrTimeout, (int, float)):
            fut: Awaitable[None] = asyncio.ensure_future(
                asyncio.sleep(selectorOrFunctionOrTimeout))
            return fut
        if not isinstance(selectorOrFunctionOrTimeout, str):
            fut = asyncio.get_event_loop().create_future()
            fut.set_exception(TypeError(
                'Unsupported target type: ' +
                str(type(selectorOrFunctionOrTimeout))
            ))
            return fut
        if ('=>' in selectorOrFunctionOrTimeout or
                selectorOrFunctionOrTimeout.strip().startswith('function')):
            return self.waitForFunction(selectorOrFunctionOrTimeout, options)
        return self.waitForSelector(selectorOrFunctionOrTimeout, options)

def __getitem__(
        self, name: Union[str, Tuple[str, bool]]) -> Callable[
            ..., Awaitable[str]]:
        if isinstance(name, tuple):
            block_name, defined_here = name

        else:
            block_name = name
            defined_here = False

        if block_name not in self._blocks.keys():
            raise KeyError(f"Unknown Block Name {block_name}.")

        SelectedBlockRuntime = self._blocks[block_name]

        async def wrapper() -> str:
            block_rt = SelectedBlockRuntime(
                self._skt_rt, _defined_here=defined_here)

            await block_rt._draw()

            return block_rt._block_result

        return wrapper

def _validate_response(cls, response: aiohttp.client.ClientResponse) -> Awaitable[aiohttp.client.ClientResponse]:
        """
        Takes in a HTTP response, looks through it to see if its legit, if not raise some errors.
        If all is good return the response

        :param response: aiohttp response
        :return: aiohttp response
        """
        if 400 <= response.status < 600:
            if response.headers.get('Content-Type') == 'application/json':
                json_data = await response.json()
                cls._raise_error(response.status, errors=json_data.get('errors'))
            else:
                text = await response.text()
                cls._raise_error(response.status, message=text)
        else:
            return response

def __init__(self, json_dict: dict, request_func: Callable[[str, Union[str, List[str]], Optional[dict], Optional[int]],
                                                               Awaitable[aiohttp.client.ClientResponse]]) -> None:
        self._request = request_func

        self.warnings = None
        self.auth = None
        self.renewable = None
        self.lease_duration = None
        self.data = None
        self.wrap_info = None

        self.lease_id = None
        self.request_id = None

        self.wrapped_at = None
        self.expires_at = None

        self._set(json_dict)

def wait(self,
                   timeout: Union[float, int],
                   func: Callable[[], Awaitable[Any]],
                   *exceptions: Exception) -> Any:
        deadline = time.time() + timeout
        err = None
        while deadline > time.time():
            try:
                result = await func()
                if result:
                    return result
                else:
                    await asyncio.sleep(0.2)
            except exceptions as exc:
                err = exc
                await asyncio.sleep(0.2)
        raise ArsenicTimeout() from err

def schedule_handler(cls: Any, obj: Any, context: Dict, func: Any, interval: Optional[Union[str, int]]=None, timestamp: Optional[str]=None, timezone: Optional[str]=None) -> Any:
        async def handler() -> None:
            values = inspect.getfullargspec(func)
            kwargs = {k: values.defaults[i] for i, k in enumerate(values.args[len(values.args) - len(values.defaults):])} if values.defaults else {}
            routine = func(*(obj,), **kwargs)
            try:
                if isinstance(routine, Awaitable):
                    await routine
            except Exception as e:
                pass

        context['_schedule_scheduled_functions'] = context.get('_schedule_scheduled_functions', [])
        context['_schedule_scheduled_functions'].append((interval, timestamp, timezone, func, handler))

        start_func = cls.start_scheduler(cls, obj, context)
        return (await start_func) if start_func else None

def __init__(self,
                 name: str,
                 request: Type[Req],
                 response: Type[Res],
                 service: Type[S],
                 implementation: Callable[[S, Req], Awaitable[Res]],
                 *,
                 http_path: str = None,
                 http_method: HTTPVerb = None,
                 http_status: int = None,
                 **options: Dict[str, Any]) -> None:
        super().__init__(name,
                         request,
                         response,
                         service,
                         http_path=http_path,
                         http_method=http_method,
                         http_status=http_status,
                         **options)
        self.implementation = implementation

def stop(self) -> Awaitable:
        """Stop."""
        contentPromise = asyncio.get_event_loop().create_future()
        self._client.once(
            'Tracing.tracingComplete',
            lambda event: asyncio.ensure_future(
                self._readStream(event.get('stream'), self._path)
            ).add_done_callback(
                lambda fut: contentPromise.set_result(
                    fut.result())  # type: ignore
            )
        )
        await self._client.send('Tracing.end')
        self._recording = False
        return await contentPromise

def waitForSelector(self, selector: str, options: dict = None,
                        **kwargs: Any) -> Awaitable:
        """Wait for selector matches element."""
        if options is None:
            options = dict()
        options.update(kwargs)
        timeout = options.get('timeout', 30_000)  # msec
        interval = options.get('interval', 0)  # msec
        return WaitTask(self, 'selector', selector, timeout, interval=interval)

def waitForFunction(self, pageFunction: str, options: dict = None,
                        *args: str, **kwargs: Any) -> Awaitable:
        """Wait for js function return true."""
        if options is None:
            options = dict()
        options.update(kwargs)
        timeout = options.get('timeout',  30_000)  # msec
        interval = options.get('interval', 0)  # msec
        return WaitTask(self, 'function', pageFunction, timeout, *args,
                        interval=interval)

def send(self, method: str, params: dict) -> Awaitable:
        """Send message via the connection."""
        self._lastId += 1
        _id = self._lastId
        msg = json.dumps(dict(
            id=_id,
            method=method,
            params=params,
        ))
        logger.debug(f'SEND?: {msg}')
        asyncio.ensure_future(self._async_send(msg))
        callback = asyncio.get_event_loop().create_future()
        self._callbacks[_id] = callback
        callback.method = method  # type: ignore
        return callback

def waitFor(self, selectorOrFunctionOrTimeout: Union[str, int, float],
                options: dict = None, **kwargs: Any) -> Awaitable:
        """Wait for function, timeout, or element which matches on page."""
        frame = self.mainFrame
        if not frame:
            raise PageError('no main frame.')
        return frame.waitFor(selectorOrFunctionOrTimeout, options, **kwargs)

def waitForSelector(self, selector: str, options: dict = None,
                        **kwargs: Any) -> Awaitable:
        """Wait until element which matches selector appears on page."""
        frame = self.mainFrame
        if not frame:
            raise PageError('no main frame.')
        return frame.waitForSelector(selector, options, **kwargs)

def waitForFunction(self, pageFunction: str, options: dict = None,
                        *args: str, **kwargs: Any) -> Awaitable:
        """Wait for function."""
        frame = self.mainFrame
        if not frame:
            raise PageError('no main frame.')
        return frame.waitForFunction(pageFunction, options, *args, **kwargs)

def _wrap_coro_func(coro: Callable[[Event], Awaitable]
                    ) -> Callable[[Event], Awaitable]:
    def wrapper(e: Event) -> Future:
        return ensure_future(coro(e))
    return wrapper

def __call__(self, event: Event) -> Awaitable[None]:
        """Execute wrapped event listener.

        Pass event object to the listener as a first argument.
        """
        return self.action(event)

def js_query(self, query: str) -> Awaitable:
        """Send query to related DOM on browser.

        :param str query: single string which indicates query type.
        """
        if self.connected:
            self.js_exec(query, self.__reqid)
            fut = Future()  # type: Future[str]
            self.__tasks[self.__reqid] = fut
            self.__reqid += 1
            return fut
        f = Future()  # type: Future[None]
        f.set_result(None)
        return f

def scrollX(self) -> Awaitable:  # noqa: D102
        return self.js_query('scrollX')

def scrollY(self) -> Awaitable:  # noqa: D102
        return self.js_query('scrollY')

def __getattr__(self, name: str) -> Callable[..., Awaitable[str]]:
        try:
            return self[name]

        except KeyError as e:
            raise AttributeError from e

def process(self, job: Job) -> Awaitable:
        self.log.debug(f'creating subprocess for {job}')
        return asyncio.create_subprocess_exec(
            job.exe,
            *job.args,
            stdin=PIPE,
            stdout=PIPE,
            stderr=PIPE,
            cwd=str(job.cwd),
            loop=self.loop,
            **job.kw,
        )

def run_coro_when_free(self, coro: Awaitable) -> None:
        while self._loop.is_running():
            time.sleep(.01)
        self._loop.run_until_complete(coro)

def tpe(self) -> Type[Awaitable]:
        return Awaitable

def extract(self, data: Awaitable, tail: List[TransEffect], in_state: bool) -> Either[R, N]:
        async def coro_map(run: Callable[[R], TransStep]) -> TransStep:
            res = await data
            return lift(run(res), in_state)
        coro = cont(tail, False, coro_map) | data
        return Lift(Propagate.one(CoroutineAlg(coro).pub))

def _get(self, path: Union[str, List[str]], params: Optional[dict] = None,
                   wrap_ttl: Optional[int] = None) -> Awaitable[aiohttp.client.ClientResponse]:
        """
        HTTP GET request

        :param path: Path components
        :param wrap_ttl: Optional TTL
        :return: A response object from aiohttp
        """
        return await self._request('get', path, payload=None, params=params, wrap_ttl=wrap_ttl)

def _delete(self, path: Union[str, List[str]], params: Optional[dict] = None,
                      wrap_ttl: Optional[int] = None) -> Awaitable[aiohttp.client.ClientResponse]:
        """
        HTTP DELETE request

        :param path: Path components
        :param wrap_ttl: Optional TTL
        :return: A response object from aiohttp
        """
        return await self._request('delete', path, payload=None, params=params, wrap_ttl=wrap_ttl)

def _list(self, path: Union[str, List[str]], params: Optional[dict] = None,
                    wrap_ttl: Optional[int] = None) -> Awaitable[aiohttp.client.ClientResponse]:
        """
        HTTP LIST request

        :param path: Path components
        :param wrap_ttl: Optional TTL
        :return: A response object from aiohttp
        """
        return await self._request('list', path, payload=None, params=params, wrap_ttl=wrap_ttl)

def _post(self, path: Union[str, List[str]], payload: Optional[dict] = None, params: Optional[dict] = None,
                    wrap_ttl: Optional[int] = None) -> Awaitable[aiohttp.client.ClientResponse]:
        """
        HTTP POST request

        :param path: Path components
        :param payload: Dictonary of key value to be turned into JSON
        :param wrap_ttl: Optional TTL
        :return: A response object from aiohttp
        """
        return await self._request('post', path, payload=payload, params=params, wrap_ttl=wrap_ttl)

def _put(self, path: Union[str, List[str]], payload: Optional[dict] = None, params: Optional[dict] = None,
                   wrap_ttl: Optional[int] = None) -> Awaitable[aiohttp.client.ClientResponse]:
        """
        HTTP PUT request

        :param path: Path components
        :param payload: Dictonary of key value to be turned into JSON
        :param wrap_ttl: Optional TTL
        :return: A response object from aiohttp
        """
        return await self._request('put', path, payload=payload, params=params, wrap_ttl=wrap_ttl)

def __init__(self, coro: typing.Awaitable[typing.List[typing.Any]]):
        self.coro = coro

        self.items = collections.deque()

        self._filled = False

def handler(create_client: Callable[[], aiohttp.ClientSession], server: ni_abc.ServerHost,
            cla_records: ni_abc.CLAHost) -> Callable[[web.Request], Awaitable[web.Response]]:
    """Create a closure to handle requests from the contribution host."""
    async def respond(request: web.Request) -> web.Response:
        """Handle a webhook trigger from the contribution host."""
        async with create_client() as client:
            try:
                contribution = await ContribHost.process(server, request, client)
                usernames = await contribution.usernames()
                server.log("Usernames: " + str(usernames))
                trusted_users = server.trusted_users()
                usernames_to_check = usernames - trusted_users
                cla_status = await cla_records.check(client, usernames_to_check)
                server.log("CLA status: " + str(cla_status))
                # With a work queue, one could make the updating of the
                # contribution a work item and return an HTTP 202 response.
                await contribution.update(cla_status)
                return web.Response(status=http.HTTPStatus.OK)
            except ni_abc.ResponseExit as exc:
                return exc.response
            except Exception as exc:
                server.log_exception(exc)
                return web.Response(
                        status=http.HTTPStatus.INTERNAL_SERVER_ERROR)

    return respond

def fetch(self, cache_only=True) \
            -> 'Union[Awaitable[ResourceObject], ResourceObject]':
        if self.session.enable_async:
            return self.fetch_async(cache_only)
        else:
            return self.fetch_sync(cache_only)

def commit(self, custom_url: str = '', meta: dict = None) \
            -> 'Union[None, ResourceObject, Awaitable[Optional[ResourceObject]]':
        """
        Commit (PATCH/POST) this resource to server.

        :param custom_url: Use this url instead of automatically determined one.
        :param meta: Optional metadata that is passed to server in POST/PATCH request

        If in async mode, this needs to be awaited.
        """
        if self.session.enable_async:
            return self._commit_async(custom_url, meta)
        else:
            return self._commit_sync(custom_url, meta)

def filter(self, filter: 'Filter') -> 'Union[Awaitable[Document], Document]':
        """
        Receive filtered list of resources. Use Filter instance.

        If in async mode, this needs to be awaited.
        """
        if self.session.enable_async:
            return self._filter_async(filter)
        else:
            return self._filter_sync(filter)

def fetch(self) -> 'Union[Awaitable[List[ResourceObject]], List[ResourceObject]]':
        """
        Fetch ResourceObjects. In practice this needs to be used only if in async mode
        and then this needs to be awaited.

        In blocking (sync) mode this is called automatically when .resource or
        .resources is accessed.
        """
        if self.session.enable_async:
            return self._fetch_async()
        else:
            return self._fetch_sync()

def execute_gen(gen: Awaitable[T]) -> T:
    """It's useful, especially in the context of scripts and tests, so be able to
    synchronous execute async functions. This is a convenience for doing that.
    """
    loop = asyncio.new_event_loop()
    result = loop.run_until_complete(gen)
    loop.close()
    return result

def async_zip(keys: Iterable[str], coros: Iterable[Awaitable[T]]) -> Dict[str, T]:
    return await async_dict(dict(zip(keys, coros)))

def async_dict(coro_dict: Dict[str, Awaitable[T]]) -> Dict[str, T]:
    keys = list(coro_dict.keys())
    results = await async_list(list(coro_dict.values()))
    return OrderedDict(zip(keys, results))

def async_list(coros: List[Awaitable[T]]) -> List[T]:
    """Use to await a list and return a list.
    Example: list_of_results = await async_list(list_of_gens)
    """
    return await asyncio.gather(*coros)

def async_tuple(*coros: Awaitable) -> Tuple[Any, ...]:
    """Await on a parameters and get a tuple back.
    Example: result_one, result_two = await async_tuple(gen_one(), gen_two())
    """
    return tuple(await asyncio.gather(*coros))

def connect(self):
        try:
            await self._connect()
        except Exception as exc:
            raise ConnectionError()
        # run any user callbacks. right now the only internal callback
        # is for pubsub channel/pattern resubscription
        for callback in self._connect_callbacks:
            task = callback(self)
            if isinstance(task, typing.Awaitable):
                await task

def only(*channels: Union[Type[DM], str], error: Optional[str]=None)\
        -> Callable[[Any, Event], Awaitable[bool]]:
    """Mark channel to allow to use handler."""

    allow_dm = False
    if DM in channels:
        channels = tuple(x for x in channels if x is not DM)
        allow_dm = True

    async def callback(bot, event: Event) -> bool:
        if isinstance(event.channel, (PrivateChannel, PublicChannel)):
            if event.channel.name in channels:
                return True
            else:
                if error:
                    await bot.say(
                        event.channel,
                        error
                    )
                return False

        if allow_dm:
            return True
        else:
            if error:
                await bot.say(
                    event.channel,
                    error
                )
            return False

    return callback

def not_(*channels: Union[Type[DM], str], error: Optional[str]=None) \
        -> Callable[[Any, Event], Awaitable[bool]]:
    """Mark channel to deny to use handler."""

    deny_dm = False
    if DM in channels:
        channels = tuple(x for x in channels if x is not DM)
        deny_dm = True

    async def callback(bot, event: Event) -> bool:
        if isinstance(event.channel, (PrivateChannel, PublicChannel)):
            if event.channel.name in channels:
                if error:
                    await bot.say(
                        event.channel,
                        error
                    )
                return False
            else:
                return True

        if deny_dm:
            if error:
                await bot.say(
                    event.channel,
                    error
                )
            return False
        else:
            return True

    return callback

def __get__(self, instance: Service, owner: Type[Service] = None) -> 'Callable[[Any, Req], Awaitable[Res]]':
        pass

    # XXX MethodDescriptor.__get__() is not used anymore

def __get__(self, instance: S, owner: Type[S] = None) -> 'Callable[[S, Req], Awaitable[Res]]':
        pass

def unfurl_image_url(url: str) -> Awaitable[str]:
    with DefaultClientSession() as session:
        results = await fetch_all(session, url)
        if 'twitter_cards' in results and 'image' in results['twitter_cards']:
            return results['twitter_cards']['image']
        if 'open_graph' in results and 'image' in results['open_graph']:
            return results['open_graph']['image']
        if 'oembed' in results and 'thumbnail_url' in results['oembed']:
            return results['oembed']['thumbnail_url']
        raise CommandError("Couldn't extract an image from the URL '{}'".format(url))

def read(self) -> Awaitable[bytes]:
        """
        Return the bytes of the file.

        Returns
        -------
        Awaitable[bytes]
            The file's data

        """
        raise NotImplemented()

def __init__(self, *, loop, request_handler: Awaitable,
                 log=None, signal=None, connections=set(), request_timeout=60,
                 request_max_size=None, has_log=True,
                 keep_alive=True, netlog=None):
        '''signal is shared'''
        self.loop = loop
        self.transport = None
        self.parser = None
        self.url = None
        self.headers = None
        self.body_channel = None
        self.message = None
        self.signal = signal
        self.has_log = has_log
        self.log = log
        self.netlog = netlog
        self.connections = connections
        self.request_handler = request_handler
        self.request_timeout = request_timeout
        self.request_max_size = request_max_size
        self._total_request_size = 0
        self._timeout_handler = None
        self._last_request_time = None
        self._request_handler_task = None
        self._request_stream_task = None
        self._is_upgrade = False
        # config.KEEP_ALIVE or not check_headers()['connection_close']
        self._keep_alive = keep_alive

def unsafe_await(self) -> "'Maybe[Awaitable]'":
        if self.is_just:
            ret = await cast(Callable[[], Awaitable], self._get)()
            return Maybe(ret)
        else:
            return cast(Maybe[Awaitable], self)

def coro(self) -> Awaitable[Either[IOException, A]]:
        async def coro() -> Either[IOException, A]:
            return self.attempt
        return coro()