Asynchronously scrape fabric information

Question

This is my first asyncio/aiohttp web scraper. I am trying to wrap my head around Python's asyncio/aiohttp libs these days and I am not sure if I fully understand it or not yet. So I'd like have some constructive enhancement reviews here.

I'm scraping Spoonflower, which contains some public API's for design data and pricing per fabric type data. My challenge was to get the design name, creator name and price of each design as per fabric type. Design name and creator name comes from this endpoint

https://pythias.spoonflower.com/search/v1/designs?lang=en&page_offset=0&sort=bestSelling&product=Fabric&forSale=true&showMatureContent=false&page_locale=en

and other pricing per fabric type data coming from this endpoint.

https://api-gateway.spoonflower.com/alpenrose/pricing/fabrics/FABRIC_'+%20fab_type%20+'?quantity=1&shipping_country=PK&currency=EUR&measurement_system=METRIC&design_id='%20str(item['designId'])%20'&page_locale=en

Each page has 84 items and 24 fabric types. I'm first getting all the names of the fabric types and storing in a list. I can loop through it and change the url dynamically then extracting designName and screenName from design page and finally extracting the price data.

Here is my code:

import asyncio
import aiohttp
import json
import requests
from bs4 import BeautifulSoup
from collections import OrderedDict


item_endpoint = 'https://pythias.spoonflower.com/search/v1/designs?lang=en&page_offset=0&sort=bestSelling&product=Fabric&forSale=true&showMatureContent=false&page_locale=en'

def get_fabric_names():
    res = requests.get('https://www.spoonflower.com/spoonflower_fabrics')
    soup = BeautifulSoup(res.text, 'lxml')
    fabrics = [fabric.find('h2').text.strip() for fabric in soup.find_all('div', {'class': 'product_detail medium_text'})]
    fabric = [("_".join(fab.upper().replace(u"\u2122", '').split())) for fab in fabrics]
    for index in range(len(fabric)):
        if 'COTTON_LAWN_(BETA)' in fabric[index]:
            fabric[index] = 'COTTON_LAWN_APPAREL'
        elif 'COTTON_POPLIN' in fabric[index]:
            fabric[index] = 'COTTON_POPLIN_BRAVA'
        elif 'ORGANIC_COTTON_KNIT' in fabric[index]:
            fabric[index] = 'ORGANIC_COTTON_KNIT_PRIMA'
        elif 'PERFORMANCE_PIQUÉ' in fabric[index]:
            fabric[index] = 'PERFORMANCE_PIQUE'
        elif 'CYPRESS_COTTON' in fabric[index]:
            fabric[index] = 'CYPRESS_COTTON_BRAVA'
    return fabric

async def fetch_design_endpoint(session, design_url):
    async with session.get(design_url) as response:
        extracting_endpoint = await response.text()
        _json_object = json.loads(extracting_endpoint)
        return _json_object['page_results']

async def fetch_pricing_data(session, pricing_endpoint):
    async with session.get(pricing_endpoint) as response:
        data_endpoint = await response.text()
        _json_object = json.loads(data_endpoint)
        items_dict = OrderedDict()
        for item in await fetch_design_endpoint(session, item_endpoint):
            designName = item['name']
            screenName = item['user']['screenName']
            fabric_name = _json_object['data']['fabric_code']
            try:
                test_swatch_meter = _json_object['data']['pricing']['TEST_SWATCH_METER']['price']
            except:
                test_swatch_meter = 'N/A'
            try:
                fat_quarter_meter = _json_object['data']['pricing']['FAT_QUARTER_METER']['price']
            except:
                fat_quarter_meter = 'N/A'
            try:
                meter = _json_object['data']['pricing']['METER']['price']
            except:
                meter = 'N/A'

            
            #print(designName, screenName, fabric_name, test_swatch_meter,fat_quarter_meter, meter)

            if (designName, screenName) not in items_dict.keys():
                items_dict[(designName, screenName)] = {}
            itemCount = len(items_dict[(designName, screenName)].values()) / 4
            return items_dict[(designName, screenName)].update({'fabric_name_%02d' %itemCount: fabric_name,
            'test_swatch_meter_%02d' %itemCount: test_swatch_meter,
            'fat_quarter_meter_%02d' %itemCount: fat_quarter_meter,
            'meter_%02d' %itemCount: meter})
                

        

async def main():
    tasks = []
    async with aiohttp.ClientSession() as session:
        fabric_type = get_fabric_names()
        design_page = await fetch_design_endpoint(session, item_endpoint)
        for item in design_page:
            for fab_type in fabric_type[0:-3]:
                pricing_url = 'https://api-gateway.spoonflower.com/alpenrose/pricing/fabrics/FABRIC_'+ fab_type +'?quantity=1&shipping_country=PK&currency=EUR&measurement_system=METRIC&design_id='+str(item['designId'])+'&page_locale=en'
                print(pricing_url)
                await fetch_pricing_data(session, pricing_url)

                tasks.append(asyncio.create_task(
                    fetch_pricing_data(session, pricing_url)

                    )
                )

        content = await asyncio.gather(*tasks)
        return content
results = asyncio.run(main())
print(results)

Any ideas and suggestions are welcome to make this scraper more Pythonic and smart.

Answer 1

While you can put request parameters in the URI directly, it's usually neater to pass those using the params argument of session.get. So for example pricing_url could just be "https://api-gateway.spoonflower.com/alpenrose/pricing/fabrics/FABRIC_" + fab_type and you could pass the rest as params={"quantity": 1, "shipping_country": "PK", "currency": "EUR", ...}). That's usually more readable.

get_fabric_names is very different than the others - it uses requests rather than aiohttp and it hardcodes its URI instead of taking it as a parameter. I'm not saying that's a problem, but it feels a bit inconsistent.

aiohttp has JSON parsing built into its responses. If all you're going to do with some_response.text() is pass it to json.loads, you may be better off just calling some_response.json() and drop the data_endpoint and extracting_endpoint variables.

I find it a bit strange that, even though there's a public-facing API, the way you obtain a list of fabric names is by extracting it from an HTML document. Now, I don't know Spoonflower's API, maybe that's the best (or only) way available, but it feels like there should be a better option.

Is the return from fetch_pricing_data really supposed to be inside the loop? It doesn't look like there are any continues or other ways to skip it, so the loop will only run for one iteration, right? It seems a bit strange to use a loop in that case.

The output format of fetch_pricing_data is a bit unusual. If I'm understanding this correctly, each fabric type gets a dict shaped like:

# I have no idea what actual values you'll work with, so I'm making some up
{
    'fabric_name_00': 'some name',
    'fabric_name_01': 'some other name',
    'test_swatch_meter_00': 20,
    'test_swatch_meter_01': 20,
    'fat_quarter_meter_00': 25,
    'fat_quarter_meter_01': 25,
    'meter_00': 5,
    'meter_01': 6,
    ...: ...
}

...which seems really awkward to me. If each number will contain all of those (which you seem to assume), I feel like an easier-to-work-with format could be:

[
    {
        'fabric_name': 'some name',
        'test_swatch_meter': 20,
        'fat_quarter_meter': 25,
        'meter': 5
    },
    {
        'fabric_name': 'some other name',
        'test_swatch_meter': 20,
        'fat_quarter_meter': 25,
        'meter': 6
    },
    ...
]

The manual replacement of fabric names in get_fabric_names feels a bit clunky. Now, depending on how the API is designed that might be the only way to actually get those fabric IDs, but even if that's the case I'd think it best to keep the list of replacements elsewhere - it looks like the kind of thing that might need changing over time, so putting it in a variable somewhere or even in a separate data file might make sense.

It's not at all clear why you skip the last 3 values in fabric_type. If it's because they aren't fabric types, I think it should be get_fabric_names's responsibility to not return them in the first place.

Finally, some nitpicks about variable names:

• fabric is a bit of a weird name considering it's a list of multiple fabrics. Maybe fabric_names? Or maybe call it fabrics and rename the existing fabrics to something like fabric_headers? Or both!
• Same for fabric_type in main - why not fabric_types?
• Most of your variables called *_endpoint are URIs you can send requests to. Clear and consistent. But data_endpoint and extracting_endpoint are responses to queries, which is mildly confusing on its own and also goes against the convention you establish.
• The name items_dict is a bit redundant. Isn't items enough?
• You seem to pretty consistently use this_casing_scheme for variable names (as is conventional in Python), except for designName, screenName and itemCount. You might want to rename those for consistency
• Why the leading _ in _json_object?

Answer 2

Something not included in the other answers:

These two lines in your main function mean that you request and parse each page twice. It also means your code is not really async, because you loop over all fabrics and just await each of them. You should be able to just remove the first one.

await fetch_pricing_data(session, pricing_url)
tasks.append(asyncio.create_task(fetch_pricing_data(session, pricing_url))

In addition, you could spawn async tasks in that function instead of the simple for loop you currently have:

for item in await fetch_design_endpoint(session, item_endpoint)

This will speed up your code by quite a lot, probably, once you have fixed the return value of that function to be outside the loop.

However, your function fetch_design_endpoint does not actually depend on anything specific in fetch_pricing_data. This is either a bug (did you want to change the URL?) or you could retrieve the results only once.

Answer 3

I'm not too experienced with asyncio, but I have a couple of minor observations about your program.

It is a code smell in modern Python to use an int to index over an array. Instead of

for index in range(len(fabric)):

You could say

for fabric_type in fabric:

Another observation in that you are using naked except blocks instead of catching "KeyError" explicitly. If you do so you save the reader some mental effort going back to see why you would be expecting a problem to be raised.