FM / asyncio

/opt/imh/python3.13/lib/python3.13/asyncio/

/opt/imh/python3.13/lib/python3.13/asyncio

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
base_events.py	80760	0644	edit dl rm
base_futures.py	1974	0644	edit dl rm
base_subprocess.py	9677	0644	edit dl rm
base_tasks.py	2672	0644	edit dl rm
constants.py	1413	0644	edit dl rm
coroutines.py	3342	0644	edit dl rm
events.py	29786	0644	edit dl rm
exceptions.py	1752	0644	edit dl rm
format_helpers.py	2727	0644	edit dl rm
futures.py	14157	0644	edit dl rm
locks.py	20580	0644	edit dl rm
log.py	124	0644	edit dl rm
mixins.py	481	0644	edit dl rm
proactor_events.py	33529	0644	edit dl rm
protocols.py	6957	0644	edit dl rm
queues.py	10152	0644	edit dl rm
runners.py	7230	0644	edit dl rm
selector_events.py	48226	0644	edit dl rm
sslproto.py	31869	0644	edit dl rm
staggered.py	7077	0644	edit dl rm
streams.py	28481	0644	edit dl rm
subprocess.py	7737	0644	edit dl rm
taskgroups.py	10049	0644	edit dl rm
tasks.py	39757	0644	edit dl rm
threads.py	790	0644	edit dl rm
timeouts.py	6062	0644	edit dl rm
transports.py	10808	0644	edit dl rm
trsock.py	2475	0644	edit dl rm
unix_events.py	54416	0644	edit dl rm
windows_events.py	32630	0644	edit dl rm
windows_utils.py	5060	0644	edit dl rm
__init__.py	1220	0644	edit dl rm
__main__.py	6128	0644	edit dl rm

Edit: /opt/imh/python3.13/lib/python3.13/asyncio/format_helpers.py (2727B)

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, *, debug=False):
    func_repr = _format_callback(func, args, None, debug=debug)
    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, *, debug=False):
    """Format function arguments and keyword arguments.

Special case for a single parameter: ('hello',) is formatted as ('hello').

Note that this function only returns argument details when
    debug=True is specified, as arguments may contain sensitive
    information.
    """
    if not debug:
        return '()'

# 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, *, debug=False, suffix=''):
    if isinstance(func, functools.partial):
        suffix = _format_args_and_kwargs(args, kwargs, debug=debug) + suffix
        return _format_callback(func.func, func.args, func.keywords,
                                debug=debug, suffix=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, debug=debug)
    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, *, debug=False): func_repr = _format_callback(func, args, None, debug=debug) 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, *, debug=False): """Format function arguments and keyword arguments. Special case for a single parameter: ('hello',) is formatted as ('hello'). Note that this function only returns argument details when debug=True is specified, as arguments may contain sensitive information. """ if not debug: return '()' # 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, *, debug=False, suffix=''): if isinstance(func, functools.partial): suffix = _format_args_and_kwargs(args, kwargs, debug=debug) + suffix return _format_callback(func.func, func.args, func.keywords, debug=debug, suffix=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, debug=debug) 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