<commit_before>from __future__ import absolute_import

import functools
import inspect

__all__ = [
    'apply_ctx',
    'sse_event',
]


def _sse_encode(k, v):
    return ''.join('%s: %s\n' % (k, line) for line in v.split('\n'))


def sse_event(event=None, data=None, id=None, retry=None, comment=None,
        encoding='utf-8'):
    """Encode a Server-Sent Event (SSE).

    At least one field must be present. All fields are strings, except
    retry, which should be an integer. The data and comment fields can contain
    newlines.
    """
    if all(x is None for x in [event, data, id, retry, comment]):
        raise TypeError("Event must have at least one field")
    return ''.join([
        _sse_encode('', comment) if comment is not None else '',
        _sse_encode('id', id) if id is not None else '',
        _sse_encode('event', event) if event is not None else '',
        _sse_encode('retry', str(retry)) if retry is not None else '',
        _sse_encode('data', data) if data is not None else '',
        '\n',
    ]).encode(encoding)


def dual_use_decorator(fn):
    """Turn a function into a decorator that can be called with or without
    arguments."""
    @functools.wraps(fn)
    def decorator(*args, **kw):
        if len(args) == 1 and not kw and callable(args[0]):
            return fn()(args[0])
        else:
            return fn(*args, **kw)
    return decorator


def dual_use_decorator_method(fn):
    """Turn a method into a decorator that can be called with or without
    arguments. """
    @functools.wraps(fn)
    def decorator(*args, **kw):
        if len(args) == 2 and not kw and callable(args[1]):
            return fn(args[0])(args[1])
        else:
            return fn(*args, **kw)
    return decorator


def get_args(obj):
    """Get a list of argument names for a callable."""
    if inspect.isfunction(obj):
        return inspect.getargspec(obj).args
    elif inspect.ismethod(obj):
        return inspect.getargspec(obj).args[1:]
    elif inspect.isclass(obj):
        return inspect.getargspec(obj.__init__).args[1:]
    elif hasattr(obj, '__call__'):
        return inspect.getargspec(obj.__call__).args[1:]
    else:
        raise TypeError("Can't inspect signature of '%s' object." % obj)


def apply_ctx(fn, ctx):
    """Return fn with ctx partially applied, if requested.

    If the `fn` callable accepts an argument named "ctx", returns a
    functools.partial object with ctx=ctx applied, else returns `fn` unchanged.
    """
    if 'ctx' in get_args(fn):
        return functools.partial(fn, ctx=ctx)
    else:
        return fn
<commit_msg>Document 'ctx' argument injection gotcha.<commit_after>from __future__ import absolute_import

import functools
import inspect

__all__ = [
    'apply_ctx',
    'sse_event',
]


def _sse_encode(k, v):
    return ''.join('%s: %s\n' % (k, line) for line in v.split('\n'))


def sse_event(event=None, data=None, id=None, retry=None, comment=None,
        encoding='utf-8'):
    """Encode a Server-Sent Event (SSE).

    At least one field must be present. All fields are strings, except
    retry, which should be an integer. The data and comment fields can contain
    newlines.
    """
    if all(x is None for x in [event, data, id, retry, comment]):
        raise TypeError("Event must have at least one field")
    return ''.join([
        _sse_encode('', comment) if comment is not None else '',
        _sse_encode('id', id) if id is not None else '',
        _sse_encode('event', event) if event is not None else '',
        _sse_encode('retry', str(retry)) if retry is not None else '',
        _sse_encode('data', data) if data is not None else '',
        '\n',
    ]).encode(encoding)


def dual_use_decorator(fn):
    """Turn a function into a decorator that can be called with or without
    arguments."""
    @functools.wraps(fn)
    def decorator(*args, **kw):
        if len(args) == 1 and not kw and callable(args[0]):
            return fn()(args[0])
        else:
            return fn(*args, **kw)
    return decorator


def dual_use_decorator_method(fn):
    """Turn a method into a decorator that can be called with or without
    arguments. """
    @functools.wraps(fn)
    def decorator(*args, **kw):
        if len(args) == 2 and not kw and callable(args[1]):
            return fn(args[0])(args[1])
        else:
            return fn(*args, **kw)
    return decorator


def get_args(obj):
    """Get a list of argument names for a callable."""
    if inspect.isfunction(obj):
        return inspect.getargspec(obj).args
    elif inspect.ismethod(obj):
        return inspect.getargspec(obj).args[1:]
    elif inspect.isclass(obj):
        return inspect.getargspec(obj.__init__).args[1:]
    elif hasattr(obj, '__call__'):
        return inspect.getargspec(obj.__call__).args[1:]
    else:
        raise TypeError("Can't inspect signature of '%s' object." % obj)


def apply_ctx(fn, ctx):
    """Return fn with ctx partially applied, if requested.

    If the `fn` callable accepts an argument named "ctx", returns a
    functools.partial object with ctx=ctx applied, else returns `fn` unchanged.

    For this to work, the 'ctx' argument must come after any arguments that are
    passed as positional arguments. For example, 'ctx' must be the 2nd argument
    for request handlers, serializers and deserializers, that are always called
    with one positional argument (the request, object to serialize, and input
    filehandle, respectively).
    """
    if 'ctx' in get_args(fn):
        return functools.partial(fn, ctx=ctx)
    else:
        return fn

{'fraction_non_alphanumeric': 0.11985688729874776, 'fraction_numerical': 0.004114490161001789, 'mean_word_length': 2.920757363253857, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4388380', 'n_tokens_mistral': 1740, 'n_tokens_neox': 1704, 'n_words': 603}