run_in_chunks.py

from typing import List, Any, Callable
import time
import functools
import csv


def run_in_chunks_with(after_chunk: Callable, size: int) -> Callable:
    """
    Function decorator. Applies to any function that
    both consumes and produces a list.

    Breaks the list into chunks of _size_ and applies the
    decorated function to each chunk, running after_chunk
    after each chunk.

    Usage:
        @run_in_chunks_with(func, 1000)
        def f(x):
            foo
    """
    def chunks(lst: List[Any], chunk_length: int) -> List[Any]:
        """Given a list and a length, yield sublists of that length"""
        for i in range(0, len(lst), chunk_length):
            yield lst[i:i+chunk_length]

    # Create inner decorator function to be wrapped
    def decorator_chunks(func: Callable) -> Callable:
        @functools.wraps(func)
        # Create inner wrapper function
        def wrapper_chunks(*args):
            lst, *_ = args
            for chunk in chunks(lst, size):
                func(chunk)
                after_chunk(chunk)
        return wrapper_chunks
    return decorator_chunks


def rate_limit(size: int, *, delay: int = 1) -> Callable:
    """
    Function decorator. Applies to any function that
    both consumes and produces a list.

    Breaks the list into chunks of _size_ and applies the
    decorated function to each chunk, waiting _delay_
    seconds between chunks.

    Usage: 1000 at a time, 1 second delay
        @run_in_chunks(1000, delay=1)
        def f(x):
            foo

    Usage: 1000 at a time, 1 minute delay
    @run_in_chunks(1000, delay=1*60)
        def f(x):
            foo
    """
    return run_in_chunks_with(lambda x: time.sleep(delay), size)


def dump_to_csv_every(size: int, *, filename: str = None) -> Callable:
    """
    Function decorator based on @run_in_chunks_with.
    Saves your work every _size_ lines.
    Has an optional argument over @run_in_chunks_with
    to specify the file to save to.

    Usage:
        @dump_to_csv_every(1000)
        def f(x):
            foo
    """
    filename = filename if filename is not None else 'temp_output.csv'

    def save(results: List[Any]):
        with open(filename, 'a') as outfile:
            writer = csv.writer(outfile, lineterminator='\n')
            writer.writerow(results)

    return run_in_chunks_with(save, size)


def main():
    """Module does nothing on its own"""
    N = 13

    @rate_limit(N, delay=0.5)
    def printer(xs: List[Any]) -> None:
        for x in xs:
            print(f'{(len(str(max(items)))-len(str(x)))*" "}{x}', end=' ')
        print('')

    items = list(range(1, 170))
    printer(items)


if __name__ == '__main__':
    main()