utils.py 文件源码

def data_mapper(mapper: Callable[[types.DataType], types.DataType]) -> Callable[[types.ListenerType], types.ListenerType]:
    '''
    Apply a mapper to the listener's `data` argument.
    '''

    def decorator(listener: types.ListenerType) -> types.ListenerType:

        @functools.wraps(listener)
        def wrapped(dtype: types.DiffType, index: int, data: types.DataType) -> None:
            listener(dtype, index, mapper(data))

        with contextlib.suppress(AttributeError):
            wrapped.finalize_batch = listener.finalize_batch

        return wrapped
    return decorator