slamdring/slamdring.py
async def slam(
    input_file,
    output_file,
    num_tasks,
    delimiter,
    format,
    request_field,
    response_field,
    repeat_request,
    ignore_exceptions
):
    """ Sets up the async queues and tasks and executes the requests.
    """
    request_queue = asyncio.Queue()
    response_queue = asyncio.Queue()

    if format == "csv":
        reader = _csv_reader(input_file, delimiter=delimiter)
        write = curry(_csv_write)(csv.writer(output_file, delimiter=delimiter))
        # If the request field is the default "request", translate that into
        # the last column. Otherwise convert the option into an integer.
        extractor = curry(_list_extractor)(
            request_field=(
                -1 if request_field == "request" else int(request_field)
            )
        )

        # Select the processor based on whether the request field is being
        # repeated, and whether the column provided for the request is a
        # negative or positive integer.
        request_field_int = -1 if request_field == "request" \
            else int(request_field)
        if not repeat_request and request_field_int < 0:
            processor = curry(_list_replace_processor_negative)(
                request_field=request_field_int
            )
        elif not repeat_request:
            processor = curry(_list_replace_processor)(
                request_field=request_field_int
            )
        else:
            processor = _list_append_processor

    elif format == "csv-header":
        reader = _csv_dict_reader(input_file, delimiter=delimiter)
        # Use the field names from the reader for the writer.
        fields = reader.fieldnames
        writer = csv.DictWriter(
            output_file,
            delimiter=delimiter,
            fieldnames=fields+[response_field]
        )
        # Write the header before doing anything else.
        writer.writeheader()
        # Construct the rest of the helpers.
        write = curry(_csv_write)(writer)
        extractor = curry(_dict_extractor)(request_field=request_field)
        processor = curry(
            _dict_replace_processor if not repeat_request
            else _dict_append_processor
        )(request_field=request_field, response_field=response_field)

    elif format == "json":
        reader = _json_reader(input_file)
        write = curry(_json_write)(output_file)
        extractor = curry(_dict_extractor)(request_field=request_field)
        processor = curry(
            _dict_replace_processor if not repeat_request
            else _dict_append_processor
        )(
            request_field=request_field,
            response_field=response_field,
            parse=_safe_json_parse
        )

    processor_tasks = [
        asyncio.ensure_future(
            process_records(
                request_queue,
                response_queue,
                extractor,
                processor,
                ignore_exceptions
            )
        )
        for ii in range(num_tasks)
    ]

    writer_task = asyncio.ensure_future(write_records(response_queue, write))

    # Lazily put the rows in the queue.
    for row in reader:
        await request_queue.put(row)

    # Wait until queues are complete.
    await request_queue.join()
    await response_queue.join()

    # Shutdown all tasks.
    for task in processor_tasks:
        task.cancel()
    writer_task.cancel()
 Sets up the async queues and tasks and executes the requests.
    
Sets up the async queues and tasks and executes the requests.
Sets up the async queues and tasks and executes the requests .
async def slam(
    input_file,
    output_file,
    num_tasks,
    delimiter,
    format,
    request_field,
    response_field,
    repeat_request,
    ignore_exceptions
):
    request_queue = asyncio.Queue()
    response_queue = asyncio.Queue()
    if format == "csv":
        reader = _csv_reader(input_file, delimiter=delimiter)
        write = curry(_csv_write)(csv.writer(output_file, delimiter=delimiter))
        extractor = curry(_list_extractor)(
            request_field=(
                -1 if request_field == "request" else int(request_field)
            )
        )
        request_field_int = -1 if request_field == "request" \
            else int(request_field)
        if not repeat_request and request_field_int < 0:
            processor = curry(_list_replace_processor_negative)(
                request_field=request_field_int
            )
        elif not repeat_request:
            processor = curry(_list_replace_processor)(
                request_field=request_field_int
            )
        else:
            processor = _list_append_processor
    elif format == "csv-header":
        reader = _csv_dict_reader(input_file, delimiter=delimiter)
        fields = reader.fieldnames
        writer = csv.DictWriter(
            output_file,
            delimiter=delimiter,
            fieldnames=fields+[response_field]
        )
        writer.writeheader()
        write = curry(_csv_write)(writer)
        extractor = curry(_dict_extractor)(request_field=request_field)
        processor = curry(
            _dict_replace_processor if not repeat_request
            else _dict_append_processor
        )(request_field=request_field, response_field=response_field)
    elif format == "json":
        reader = _json_reader(input_file)
        write = curry(_json_write)(output_file)
        extractor = curry(_dict_extractor)(request_field=request_field)
        processor = curry(
            _dict_replace_processor if not repeat_request
            else _dict_append_processor
        )(
            request_field=request_field,
            response_field=response_field,
            parse=_safe_json_parse
        )
    processor_tasks = [
        asyncio.ensure_future(
            process_records(
                request_queue,
                response_queue,
                extractor,
                processor,
                ignore_exceptions
            )
        )
        for ii in range(num_tasks)
    ]
    writer_task = asyncio.ensure_future(write_records(response_queue, write))
    for row in reader:
        await request_queue.put(row)
    await request_queue.join()
    await response_queue.join()
    for task in processor_tasks:
        task.cancel()
    writer_task.cancel()
async def slam ( input_file , output_file , num_tasks , delimiter , format , request_field , response_field , repeat_request , ignore_exceptions ) : request_queue = asyncio . Queue ( ) response_queue = asyncio . Queue ( ) if format == "csv" : reader = _csv_reader ( input_file , delimiter = delimiter ) write = curry ( _csv_write ) ( csv . writer ( output_file , delimiter = delimiter ) ) extractor = curry ( _list_extractor ) ( request_field = ( - 1 if request_field == "request" else int ( request_field ) ) ) request_field_int = - 1 if request_field == "request" else int ( request_field ) if not repeat_request and request_field_int < 0 : processor = curry ( _list_replace_processor_negative ) ( request_field = request_field_int ) elif not repeat_request : processor = curry ( _list_replace_processor ) ( request_field = request_field_int ) else : processor = _list_append_processor elif format == "csv-header" : reader = _csv_dict_reader ( input_file , delimiter = delimiter ) fields = reader . fieldnames writer = csv . DictWriter ( output_file , delimiter = delimiter , fieldnames = fields + [ response_field ] ) writer . writeheader ( ) write = curry ( _csv_write ) ( writer ) extractor = curry ( _dict_extractor ) ( request_field = request_field ) processor = curry ( _dict_replace_processor if not repeat_request else _dict_append_processor ) ( request_field = request_field , response_field = response_field ) elif format == "json" : reader = _json_reader ( input_file ) write = curry ( _json_write ) ( output_file ) extractor = curry ( _dict_extractor ) ( request_field = request_field ) processor = curry ( _dict_replace_processor if not repeat_request else _dict_append_processor ) ( request_field = request_field , response_field = response_field , parse = _safe_json_parse ) processor_tasks = [ asyncio . ensure_future ( process_records ( request_queue , response_queue , extractor , processor , ignore_exceptions ) ) for ii in range ( num_tasks ) ] writer_task = asyncio . ensure_future ( write_records ( response_queue , write ) ) for row in reader : await request_queue . put ( row ) await request_queue . join ( ) await response_queue . join ( ) for task in processor_tasks : task . cancel ( ) writer_task . cancel ( )
Sets up the async queues and tasks and executes the requests.
Sets up the async queues and tasks and executes the requests .
""" Sets up the async queues and tasks and executes the requests.
    """ # If the request field is the default "request", translate that into # the last column. Otherwise convert the option into an integer. # Select the processor based on whether the request field is being # repeated, and whether the column provided for the request is a # negative or positive integer. # Use the field names from the reader for the writer. # Write the header before doing anything else. # Construct the rest of the helpers. # Lazily put the rows in the queue. # Wait until queues are complete. # Shutdown all tasks.
input_file
output_file
num_tasks
delimiter
format
request_field
response_field
repeat_request
ignore_exceptions