6
\$\begingroup\$

I have a CSV containing ~600k part numbers to be uploaded to my website's inventory. However, this CSV only contains limited information. We're missing pricing and other related information. To get this information I'm required to make requests to the provider's API, and add it into the CSV. At the moment, I've been splitting this part-file into 6 pieces and running the script on each of these files simultaneously. If I run one script it will take hours. Whereas if I split it up, it will go considerably faster.

The process:

  1. Read Partnumber from CSV
  2. Make request
  3. If errors, continue, and notate error
  4. If inventory, write to inventory.csv with ID and warehouse info
  5. Place part info into results.csv
  6. Onto the next one

I was thinking that I could assign each item a unique ID, have the script request that information, go back into the original CSV and finally place the information back into the original document.

How can I utilize the full potential of the system I'm running this script on?

Here's what I've got so far:

import csv
import zeep

wsdl = '#####'
client = zeep.Client(wsdl=wsdl)

def get_data():
    with  open('partfile.csv')as f:
        parts = csv.reader(f, delimiter='|')
        with open('results.csv' , 'w+') as outfile:
            with open('inventory.csv', 'w+') as inventoryfile:
                output = csv.writer(outfile, delimiter=',')
                inventoryoutput = csv.writer(inventoryfile, delimiter=',')
                inventoryoutput.writerow([
                    'ID',
                    'WarehouseNumber',
                    'WarehouseName',
                    'QuantityAvailable'
                ])
                # Header Row
                output.writerow([
                    'ID',
                    'Make',
                    'Part Number',
                    'Price',
                    'Dealer Price',
                    'Retail Price',
                    'List Price',
                    'Core Cost',
                    'Part Description',
                    'Is Discontinued',
                    'Is Dropship Only',
                    'Is Refrigerant',
                    'Is Oversize',
                    'Is Hazmat',
                    'Sub Parts',
                    'Cross Reference Parts',
                    'Log',
                    'Total Inventory'
                ])
                itemId = 0
                for row in parts:
                    try:
                        item = client.service.ExactPartLookup('#####', '#####', row[0], row[1])
                        if (item == None):
                            raise Exception('Item is None')
                    except:
                        write_error(row[1])
                        continue


                    item = item.PartInformation_v2[0]
                    totalInventory = 0

                    data = [
                        itemId,
                        item.Make,
                        item.PartNumber,
                        item.Price,
                        item.Dealer,
                        item.Retail,
                        item.List,
                        item.CoreCost,
                        item.PartDescription,
                        item.IsDiscontinued,
                        item.IsDropShipOnly,
                        item.IsRefrigerant,
                        item.IsOversize,
                        item.IsHazmat,
                        item.SubParts,
                        item.CrossReferenceParts,
                        item.Log
                    ]

                    print(item.PartNumber)

                    if (item.Inventory != None):
                        inventory = item.Inventory.InventoryInformation_v2
                        iterator = 0
                        for i in inventory:
                            inventoryoutput.writerow([
                                itemId,
                                inventory[iterator].WarehouseNumber,
                                inventory[iterator].WarehouseName,
                                inventory[iterator].QuantityAvailable
                            ])
                            totalInventory += inventory[iterator].QuantityAvailable
                            iterator += 1

                    data.append(totalInventory)
                    itemId += 1
                    output.writerow(data)

def write_error( partNumber ):
    with open("errors.log", "a+") as errorfile:
        errorfile.write("Error! Part Number: " + partNumber + "\n")

get_data()
\$\endgroup\$
  • \$\begingroup\$ Can you get a bulk dump of all ExactPartNumbers from API or are you limited with LookUp service? \$\endgroup\$ – Parfait Oct 2 '16 at 23:43
  • \$\begingroup\$ Hi! Unfortunately there is no way to download a bulk dump. \$\endgroup\$ – Ryan Cady Oct 4 '16 at 13:32
1
\$\begingroup\$

For the current code:

  • The nested withs are adding unnecessary indentation, they can just be written with a single with, possibly on multiple lines:

    with ... as f, ... as outfile, ...:
    
  • The variable names should adhere to PEP8, so single characters and camelCase are discouraged.

  • x is None is slightly more idiomatic.
  • No parentheses are needed for ifs.
  • The iterator variable is interesting - unless there's a reason I don't see it should be done with regular iteration, enumerate, .items() or some other way to directly iterate over the values instead of indexing and counting up manually. Same goes for itemId, just use enumerate.
  • I'd probably keep the error file open instead of reopening it over and over. If you need to flush to see the output then do that.
  • The writerow header should probably be moved into constants so that it doesn't clutter the function so much.

Now for the performance ... there's no reason not to run this in parallel, e.g. issue the requests in separate threads (or whatever mechanism), then collect the results and write them out. If the order of entries isn't relevant it's even easier.

\$\endgroup\$
  • \$\begingroup\$ Thanks for this ferada! The order of entries definitely ISN'T important so hopefully that will make it easier. However, I'm more looking for how to parallelize this code. I'm fairly new with Python, as well as multiprocessing/multithreading. I'll keep researching! \$\endgroup\$ – Ryan Cady Oct 6 '16 at 22:36
  • \$\begingroup\$ Unless you already have take a look at the standard multiprocessing module, in particular the Pool class and the map / map_async methods. In general you could also look at asynchronous I/O, but I think that might not work so well with the zeep library. \$\endgroup\$ – ferada Oct 6 '16 at 23:05
1
\$\begingroup\$

Okay, so I got it working with multiprocessing by doing more research!

Basically I load the CSV into a list, as well as count the lines of the list. Then I can decide how many processes that I'd like to spawn. After doing that I divide the number of lines by the number of processes to determine the chunksize. Using chunksize is important because it allows me to split the iterable so that I'm not processing the same lines multiple times. Other than splitting some of the functionality into other functions and adding multiprocessing, the process is relatively identical.

Here's the result:

import csv
import itertools
import math
import zeep
import time
import json
from multiprocessing import Pool

wsdl = 'XXX'
client = zeep.Client(wsdl=wsdl)


def build_csv():
    with open('results.csv' , 'w') as outfile, open('inventory.csv', 'w') as inventoryfile:
        output = csv.writer(outfile, delimiter=',')
        inventoryoutput = csv.writer(inventoryfile, delimiter=',')
        inventoryoutput.writerow([
            'ID',
            'WarehouseNumber',
            'WarehouseName',
            'QuantityAvailable'
        ])
        output.writerow([
            'ID',
            'Make',
            'Part Number',
            'Price',
            'Dealer Price',
            'Retail Price',
            'List Price',
            'Core Cost',
            'Part Description',
            'Is Discontinued',
            'Is Dropship Only',
            'Is Refrigerant',
            'Is Oversize',
            'Is Hazmat',
            'Sub Parts',
            'Cross Reference Parts',
            'Log',
            'Total Inventory'
        ])

def get_file(filename):
    with open(filename) as file:
        row = csv.reader(file, delimiter = '|')
        row =  list( row )
        return row

def get_num_lines(filename):
    NUM_LINES = 0
    with open(filename) as file:
        row = csv.reader(file, delimiter = '|')
        for i in row:
            NUM_LINES += 1
        return NUM_LINES

def make_request(x):
    try:
        item = client.service.ExactPartLookup('XXX', 'XXX', x[0], x[1])
        if (item == None):
            write_error(x[1], 'nonetype')
        else:
            write_item(item)
    except:
        write_error(x[1], 'exception')
        pass

def write_item(item):
    with open('results.csv', 'a') as outfile, open('inventory.csv', 'a') as inventoryfile:
        inventoryoutput = csv.writer(inventoryfile, delimiter=',')
        output = csv.writer(outfile, delimiter=',')
        item = item.PartInformation_v2[0]

        itemId = item.Make + '^' + item.PartNumber

        totalInventory = 0
        data = [
            itemId,
            item.Make,
            item.PartNumber,
            item.Price,
            item.Dealer,
            item.Retail,
            item.List,
            item.CoreCost,
            item.PartDescription,
            item.IsDiscontinued,
            item.IsDropShipOnly,
            item.IsRefrigerant,
            item.IsOversize,
            item.IsHazmat,
            item.SubParts,
            item.CrossReferenceParts,
            item.Log
        ]
        if (item.Inventory != None):
            inventory = item.Inventory.InventoryInformation_v2
            iterator = 0
            for i in inventory:
                inventoryoutput.writerow([
                    itemId,
                    inventory[iterator].WarehouseNumber,
                    inventory[iterator].WarehouseName,
                    inventory[iterator].QuantityAvailable
                ])
                totalInventory += inventory[iterator].QuantityAvailable

        data.append(totalInventory)
        output.writerow(data)
        #print(item.PartNumber)

def write_error( partNumber, err_type ):
    with open("errors.log", "a+") as errorfile:
        if (err_type == 'nonetype'):
            errorfile.write("Error! NoneType... Part Number: " + partNumber + "\n")
        elif (err_type == 'exception'):
            errorfile.write("Error! Zeep Exception, Part Number: " + partNumber + "\n")

if __name__ == '__main__':
    FILE = 'PartFile.txt'
    iterable = get_file(FILE)
    NUM_LINES = get_num_lines(FILE)
    NUM_PROCS = 50
    build_csv()

    start = time.time()
    print('Starting...')

    with Pool(NUM_PROCS) as p:
        p.map(make_request, iterable, math.ceil(NUM_LINES / NUM_PROCS))

    end = time.time()
    print(end - start)
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.