diff options
author | orivej <orivej@yandex-team.ru> | 2022-02-10 16:44:49 +0300 |
---|---|---|
committer | Daniil Cherednik <dcherednik@yandex-team.ru> | 2022-02-10 16:44:49 +0300 |
commit | 718c552901d703c502ccbefdfc3c9028d608b947 (patch) | |
tree | 46534a98bbefcd7b1f3faa5b52c138ab27db75b7 /contrib/tools/python3/src/Lib/asyncio/format_helpers.py | |
parent | e9656aae26e0358d5378e5b63dcac5c8dbe0e4d0 (diff) | |
download | ydb-718c552901d703c502ccbefdfc3c9028d608b947.tar.gz |
Restoring authorship annotation for <orivej@yandex-team.ru>. Commit 1 of 2.
Diffstat (limited to 'contrib/tools/python3/src/Lib/asyncio/format_helpers.py')
-rw-r--r-- | contrib/tools/python3/src/Lib/asyncio/format_helpers.py | 152 |
1 files changed, 76 insertions, 76 deletions
diff --git a/contrib/tools/python3/src/Lib/asyncio/format_helpers.py b/contrib/tools/python3/src/Lib/asyncio/format_helpers.py index 27d11fd4fa..6618e78249 100644 --- a/contrib/tools/python3/src/Lib/asyncio/format_helpers.py +++ b/contrib/tools/python3/src/Lib/asyncio/format_helpers.py @@ -1,76 +1,76 @@ -import functools -import inspect -import reprlib -import sys -import traceback - -from . import constants - - -def _get_function_source(func): - func = inspect.unwrap(func) - if inspect.isfunction(func): - code = func.__code__ - return (code.co_filename, code.co_firstlineno) - if isinstance(func, functools.partial): - return _get_function_source(func.func) - if isinstance(func, functools.partialmethod): - return _get_function_source(func.func) - return None - - -def _format_callback_source(func, args): - func_repr = _format_callback(func, args, None) - source = _get_function_source(func) - if source: - func_repr += f' at {source[0]}:{source[1]}' - return func_repr - - -def _format_args_and_kwargs(args, kwargs): - """Format function arguments and keyword arguments. - - Special case for a single parameter: ('hello',) is formatted as ('hello'). - """ - # use reprlib to limit the length of the output - items = [] - if args: - items.extend(reprlib.repr(arg) for arg in args) - if kwargs: - items.extend(f'{k}={reprlib.repr(v)}' for k, v in kwargs.items()) - return '({})'.format(', '.join(items)) - - -def _format_callback(func, args, kwargs, suffix=''): - if isinstance(func, functools.partial): - suffix = _format_args_and_kwargs(args, kwargs) + suffix - return _format_callback(func.func, func.args, func.keywords, suffix) - - if hasattr(func, '__qualname__') and func.__qualname__: - func_repr = func.__qualname__ - elif hasattr(func, '__name__') and func.__name__: - func_repr = func.__name__ - else: - func_repr = repr(func) - - func_repr += _format_args_and_kwargs(args, kwargs) - if suffix: - func_repr += suffix - return func_repr - - -def extract_stack(f=None, limit=None): - """Replacement for traceback.extract_stack() that only does the - necessary work for asyncio debug mode. - """ - if f is None: - f = sys._getframe().f_back - if limit is None: - # Limit the amount of work to a reasonable amount, as extract_stack() - # can be called for each coroutine and future in debug mode. - limit = constants.DEBUG_STACK_DEPTH - stack = traceback.StackSummary.extract(traceback.walk_stack(f), - limit=limit, - lookup_lines=False) - stack.reverse() - return stack +import functools +import inspect +import reprlib +import sys +import traceback + +from . import constants + + +def _get_function_source(func): + func = inspect.unwrap(func) + if inspect.isfunction(func): + code = func.__code__ + return (code.co_filename, code.co_firstlineno) + if isinstance(func, functools.partial): + return _get_function_source(func.func) + if isinstance(func, functools.partialmethod): + return _get_function_source(func.func) + return None + + +def _format_callback_source(func, args): + func_repr = _format_callback(func, args, None) + source = _get_function_source(func) + if source: + func_repr += f' at {source[0]}:{source[1]}' + return func_repr + + +def _format_args_and_kwargs(args, kwargs): + """Format function arguments and keyword arguments. + + Special case for a single parameter: ('hello',) is formatted as ('hello'). + """ + # use reprlib to limit the length of the output + items = [] + if args: + items.extend(reprlib.repr(arg) for arg in args) + if kwargs: + items.extend(f'{k}={reprlib.repr(v)}' for k, v in kwargs.items()) + return '({})'.format(', '.join(items)) + + +def _format_callback(func, args, kwargs, suffix=''): + if isinstance(func, functools.partial): + suffix = _format_args_and_kwargs(args, kwargs) + suffix + return _format_callback(func.func, func.args, func.keywords, suffix) + + if hasattr(func, '__qualname__') and func.__qualname__: + func_repr = func.__qualname__ + elif hasattr(func, '__name__') and func.__name__: + func_repr = func.__name__ + else: + func_repr = repr(func) + + func_repr += _format_args_and_kwargs(args, kwargs) + if suffix: + func_repr += suffix + return func_repr + + +def extract_stack(f=None, limit=None): + """Replacement for traceback.extract_stack() that only does the + necessary work for asyncio debug mode. + """ + if f is None: + f = sys._getframe().f_back + if limit is None: + # Limit the amount of work to a reasonable amount, as extract_stack() + # can be called for each coroutine and future in debug mode. + limit = constants.DEBUG_STACK_DEPTH + stack = traceback.StackSummary.extract(traceback.walk_stack(f), + limit=limit, + lookup_lines=False) + stack.reverse() + return stack |