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

def _send_receive(self, nummsgs: int, outformat: str='json',
                      dataupdate: Optional[Dict[AnyStr, Any]]=None,
                      restart_data: bool=True) -> List[Response]:
        if restart_data:
            self._restart_data(outformat)

        if dataupdate:
            self.data.update(dataupdate)

        self._add_to_buffer(nummsgs, outformat)
        self.sendbuffer.seek(0)

        processor, _ = get_processor_instance(
                outformat,
                custom_outbuffer=self.recvbuffer,
                custom_inbuffer=self.sendbuffer
        )
        processor.process_requests(self.sendbuffer)
        return self._loadResults(outformat)

def mixtohash(self,
                  args=(),      # type: Sequence[AnyStr]
                  exe=None,     # type: Optional[str]
                  depfiles=(),  # type: Sequence[str]
                  hashobj=None  # type: Optional[Any]
                  ):
        # type: (...) -> Any
        if hashobj is None:
            hashobj = HASHFUNC()
        for filename in depfiles:
            hashobj.update(sysfilename(filename))
            hashobj.update(filesha(filename))
            hashobj.update(b'\x00')
        for arg in args:
            hashobj.update(sysfilename(arg))
            hashobj.update(b'\x00')
        if exe is not None:
            hashobj.update(self.digest_for_exe(exe))
        return hashobj

def __init__(self):
        """
        TODO
        :param program_ast: TODO
        :return: None
        """
        super(PyCoolSemanticAnalyser, self).__init__()

        # Initialize the internal program ast instance.
        self._program_ast = None

        # Classes Map: maps each class name (key: String) to its class instance (value: AST.Class).
        # Dict[AnyStr, AST.Class]
        self._classes_map = dict()

        # Class Inheritance Graph: maps a parent class (key: String) to a unique collection of its 
        #   children classes (value: set).
        # Dict[AnyStr, Set]
        self._inheritance_graph = defaultdict(set)

    # #########################################################################
    #                                PUBLIC                                   #
    # #########################################################################

def _traverse_inheritance_graph(self, starting_node: AnyStr, seen: Dict) -> bool:
        """
        Depth-First Traversal of the Inheritance Graph.
        :param starting_node: TODO
        :param seen: TODO
        :return: TODO
        """
        if seen is None:
            seen = {}

        seen[starting_node] = True

        # If the starting node is not a parent class for any child classes, then return!
        if starting_node not in self._inheritance_graph:
            return True

        # Traverse the children of the current node
        for child_node in self._inheritance_graph[starting_node]:
            self._traverse_inheritance_graph(starting_node=child_node, seen=seen)

        return True

def test_short_comment_types(create_project, run):
    """Test type hinting with short-form comments."""
    with create_project('''
        from typing import Any, AnyStr

        def types(str1, num):
            # type: (AnyStr, int) -> Any
            """usage: say types <str1> <num>"""
            print(type(str1))
            print(type(num))
    '''):
        type_reprs = run('say types world 4', stderr=True).strip().split('\n')
        assert type_reprs == [
            repr(str),
            repr(int)
        ]

def _do_post(self, url: str, *,
                       msg: Optional[Dict[AnyStr, Any]] = None,
                       token: Optional[AnyStr] = None):
        """
        Perform a POST request, validating the response code.
        This will throw a SlackAPIError, or decendent, on non-200
        status codes

        :param url: url for the request
        :param msg: payload to send
        :param token: optionally override the set token.
        :type msg: dict
        :return: Slack API Response
        :rtype: dict
        """
        msg = msg or {}
        logger.debug('Querying SLACK HTTP API: %s', url)
        msg['token'] = token or self._token
        async with self._session.post(url, data=msg) as response:
            return await self._validate_response(response, url)

def __init__(
        self,
        _=None,  # type: Optional[Union[AnyStr, typing.Mapping, typing.Sequence, typing.IO]]
    ):
        self._meta = None
        if _ is not None:
            if isinstance(_, HTTPResponse):
                meta.get(self).url = _.url
            _ = deserialize(_)
            for k, v in _.items():
                try:
                    self[k] = v
                except KeyError as e:
                    if e.args and len(e.args) == 1:
                        e.args = (
                            r'%s.%s: %s' % (type(self).__name__, e.args[0], json.dumps(_)),
                        )
                    raise e

def iprint(category, s, prefix='', end='\n', fp=None):
    # type: (str, AnyStr, str, str, Optional[IO[AnyStr]]) -> None
    category_print(args_info, 'info', category, s, prefix, end, fp=fp)

def dprint(category, s, prefix='', end='\n', fp=None):
    # type: (str, AnyStr, str, str, Optional[IO[AnyStr]]) -> None
    category_print(args_debug, 'debug', category, s, prefix, end, fp=fp)

def reporterror(s, fp=None):
    # type: (str, Optional[IO[AnyStr]]) -> None
    if fp is None:
        fp = rawstream(sys.stderr)  # type: ignore
    reportmessage(s, fp=fp)

def option_make(optionname,      # type: AnyStr
                optiontype,      # type: AnyStr
                configs,         # type: Iterable[OptionValue]
                nestedopts=None  # type: Optional[StyleDef]
                ):
    # type: (...) -> Tuple[str, str, List[OptionValue], Optional[StyleDef]]
    configs = [typeconv(c) for c in configs]
    return unistr(optionname), unistr(optiontype), configs, nestedopts

def generate(**kwargs: Any) -> Tuple[
        Dict[str, str], Union[AnyStr, aiohttp.FormData]]:
    headers = {}  # type: Dict[str, str]
    try:
        data = json.dumps(kwargs)
        headers["Content-Type"] = "application/json"

    except:  # Fallback to Form Data.
        data = _generate_form_data(**kwargs)

    return headers, data

def __init__(
        self, *args: Any, status_code: Optional[int]=None,
        content: Optional[Union[Dict[str, Any], List[Any], AnyStr]]=None,
            **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)

        self.status_code = status_code
        self.content = content

def render(self, url: str, format: str = 'html') -> AnyStr:
        self.on('Page.loadEventFired', partial(self._on_page_load_event_fired, format=format))
        self.on('Network.loadingFinished', partial(self._on_loading_finished, format=format))
        try:
            await self.navigate(url)
            self._url = url
            return await self._render_future
        finally:
            self._url = None
            self._callbacks.clear()
            self._futures.clear()
            await self._disable_events()

def mockup_http_static_server(content: bytes = b'Simple file content.', content_type: str = None, **kwargs):
    class StaticMockupHandler(BaseHTTPRequestHandler):  # pragma: no cover
        def serve_text(self):
            self.send_header('Content-Type', "text/plain")
            self.send_header('Content-Length', str(len(content)))
            self.send_header('Last-Modified', self.date_time_string())
            self.end_headers()
            self.wfile.write(content)

        def serve_static_file(self, filename: AnyStr):
            self.send_header('Content-Type', guess_type(filename))
            with open(filename, 'rb') as f:
                self.serve_stream(f)

        def serve_stream(self, stream: FileLike):
            buffer = io.BytesIO()
            self.send_header('Content-Length', str(copy_stream(stream, buffer)))
            self.end_headers()
            buffer.seek(0)
            try:
                copy_stream(buffer, self.wfile)
            except ConnectionResetError:
                pass

        # noinspection PyPep8Naming
        def do_GET(self):
            self.send_response(HTTPStatus.OK)
            if isinstance(content, bytes):
                self.serve_text()
            elif isinstance(content, str):
                self.serve_static_file(content)
            else:
                self.send_header('Content-Type', content_type)
                # noinspection PyTypeChecker
                self.serve_stream(content)

    return simple_http_server(StaticMockupHandler, **kwargs)

def test_string_types(create_project, run, type_, expected):
    """Test type hinting with string types."""
    with create_project('''
        from typing import Any, AnyStr, ByteString

        def types(value):
            # type: ({type}) -> Any
            """usage: say types <value>"""
            print(type(value))
    '''.format(type=type_)):
        assert run('say types abc').strip().split('\n') == [repr(expected)]

def create_eval_metric(metric_name: AnyStr) -> mx.metric.EvalMetric:
        """
        Creates an EvalMetric given a metric names.
        """
        # output_names refers to the list of outputs this metric should use to update itself, e.g. the softmax output
        if metric_name == C.ACCURACY:
            return utils.Accuracy(ignore_label=C.PAD_ID, output_names=[C.SOFTMAX_OUTPUT_NAME])
        elif metric_name == C.PERPLEXITY:
            return mx.metric.Perplexity(ignore_label=C.PAD_ID, output_names=[C.SOFTMAX_OUTPUT_NAME])
        else:
            raise ValueError("unknown metric name")

def create_eval_metric_composite(metric_names: List[AnyStr]) -> mx.metric.CompositeEvalMetric:
        """
        Creates a composite EvalMetric given a list of metric names.
        """
        metrics = [TrainingModel.create_eval_metric(metric_name) for metric_name in metric_names]
        return mx.metric.create(metrics)

def escape_js(value: t.AnyStr) -> str:
    """Hex encodes characters for use in JavaScript strings.

    :param value: String to be escaped.
    :return: A string safe to be included inside a <script> tag.
    """
    return str(value).translate(_js_escapes)

def filter_or_import(name: AnyStr) -> Callable:
    """
    Loads the filter from the current core or tries to import the name.

    :param name: The name to load.
    :return:  A callable.
    """
    core = get_proxy_or_core()

    try:
        ns, func = name.split(".", 1)
        return getattr(getattr(core, ns), func)
    except (ValueError, AttributeError):
        return import_item(name)

def watch(self, callback: TCallable[[AnyStr, TAny, TAny], None]) -> None:
        """
        Register a new callback that pushes or undefines the object
        from the actual environmental namespace.

        :param callback:   The callback to run
        """
        self.watchers.append(callback)

def unwatch(self, callback: TCallable[[AnyStr, TAny, TAny], None]) -> None:
        """
        Unregister a given callback

        :param callback: The callback to unregister
        """
        self.watchers.remove(callback)

def _notify(self, key: AnyStr, value: TAny, old: TAny):
        for watcher in self.watchers:
            watcher(key, value, old)

def as_dict(self) -> TDict[AnyStr, TAny]:
        return self.namespace.copy()

def __getitem__(self, item: AnyStr) -> TAny:
        return self.namespace[item]

def __setitem__(self, key: AnyStr, value: TAny):
        old = self.namespace.get(key, self.Undefined)
        self.namespace[key] = value
        self._notify(key, value, old)

def __delitem__(self, key: AnyStr):
        old = self.namespace.pop(key)
        self._notify(key, self.Undefined, old)

def _set_cb_unique(self) -> TCallable[[AnyStr, Any], None]:
        return self.push_value

def push_value(self, key: AnyStr, value: Any, old: Any) -> None:
        if value is YuunoNamespace.Undefined:
            self.environment.parent.log.debug(f"Popping from user namespace: {key}")
            self.environment.ipython.drop_by_id({key: old})
        else:
            self.environment.parent.log.debug(f"Pushing to user namespace: {key}: {value!r}")
            self.environment.ipython.push({key: value})

def test_short_form_multi():
    """Test type hinting with short-form comments and multiple args."""
    from typing import Any, AnyStr

    def func(arg1, arg2):
        # type: (AnyStr, int) -> Any
        pass

    assert get_type_hints(func, globals(), locals()) == {
        'return': Any,
        'arg1': AnyStr,
        'arg2': int
    }

def __init__(self, parent: 'BaseSprite', side: AnyStr):
        self.side = side
        self.parent = parent

def dumps(self, value: Union[AnyStr, dict]) -> bytes:
        # FIXME handle structs with bytes vals, eg, [1, '2', b'3']
        # currently self.loads(self.dumps([1, '2', b'3'])) == [1, '2', '3']
        return zlib.compress(ujson.dumps(value).encode())