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)
python类AnyStr()的实例源码
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
}
