How I Scraped Over 25,000 Forum Posts In 3 Steps

You will notice that async / await is used here.

Since I will be creating an asynchronous approach to collecting data, I don’t want my code to get blocked when working with the database. Yes, it is true that your main problem when working with asyncio is that you have to understand which operations can block your code. In addition to that, you also need to find asynchronous libraries and drivers. For MongoDB this is Motor.

Coding

What you did NOT want to know about Asyncio in web scraping

If you are parsing data from asyncio, the most important rule you should follow is REQUEST CONTROL.

Asyncio actually spawns sockets that handle your requests. That means, if you create 500 requests to one site without using await on the request function, you will open 500 sockets accessing one site. You will create 500 requests. Have you heard of DDoS attacks? Congratulations, you’ve started your journey in that direction.

Example:

for _ in range(500):
   asyncio.create_task(asyncio_requests()) # There is no await, you have just created 500 requests for the site

You could also do this

for _ in range(500):
   await asyncio.create_task(asyncio_requests()) # There is await here you make 1 request. And you wait for it to finish

You just made your code synchronous. Then why bother using asyncio at all?

This is a joke, of course, but it’s also a 100% working approach. You could parse data from multiple sites this way, at the same time creating no more than 1 request to a particular site. And it all happens in one thread. Magic.

In asyncio, it is very easy to create a situation where you generate cascade queries.

One page of topics has 30 topics. We get a total of 31 requests: for 30 topics + for the next page of topics. The site may not have time to respond to all the requests and you will end up creating another 31. Thus you will generate a growing load on the server site.

We need to remember that all resources are limited to their own extent. One site may handle no more than 100 parallel requests, the other one can handle a thousand. But you should understand that the load you create interferes with other users, or makes the server scale, and this costs money. This is why some put anti-scraping protection in place. I won’t mention any other workarounds for these protections other than using a proxy.

So, I have outlined the problem. In the Dataquest situation, you will get a notification that you are creating too many requests.

How to solve this problem in asyncio? Use Semaphore.

I created a class that doesn’t allow the creation of concurrent queries that exceed a specific limit and also caches data to a disk.

class Downloader:
   cache_path = CACHE_PATH
   attemps = 20

   def __init__(self):
       self.locker = asyncio.Semaphore(REQUESTS_LIMIT) # Creates a semaphore object
       if not self.cache_path.exists():
           self.cache_path.mkdir(parents=True)

   async def start(self):
       self.session = ClientSession(headers=HEADERS)

   async def stop(self):
       await self.session.close()

   async def request(self, method, url, *args, **kwargs):
       async with self.locker: # Here we use a semaphore. Not allowing to create queries greater than REQUESTS_LIMIT
           for _ in range(self.attemps):
               async with self.session.request(method, url, *args, **kwargs) as resp:
                   if resp.status == 200:
                       return await resp.read()
                   else:
                       await asyncio.sleep(2)

   async def get(self, *args, **kwargs):
       return await self.cache("GET", *args, **kwargs)

   async def post(self, *args, **kwargs):
       return await self.cache("POST", *args, **kwargs)

   async def cache(self, method, url, *args, **kwargs):
       url_hash = md5(url.encode()).hexdigest()
       file_name = f"{url_hash}.cache"
       file_path = Path(self.cache_path, file_name)
       if file_path.exists():
           async with aopen(file_path, 'rb') as f:
               response_data = await f.read()
           logger.info(f"Url {url} read from {file_name} cache")
       else:
           response_data = await self.request(method, url, *args, **kwargs)
           if not response_data or b"Too Many Requests" in response_data:
               raise Exception("Not Data")
           async with aopen(file_path, 'wb') as f:
               await f.write(response_data)
           logger.info(f"Url {url} cached in {file_name} cache")
       return response_data

Caching - Courtesy of the Web Parser

Why bother caching data to the disk?

Imagine you visited a page and got results. And then you noticed that you forgot to include 1 or more fields in the results. Then you need to restart the script. This takes time, creating a load on the site.

If you cache the data, instead of re-running your script on the website, you simply read data from your disk. That means the data is accessible any time, you can change result fields in the database and update the data very quickly - all of that without creating additional load for the site.

Of course, if you need to update the data regularly, you have to create a lifecycle mechanism for the cache. For example, if data has to be updated daily, then you have to remove all cache files from the previous day.

Optimization and launch

In my script, I used a limit of no more than 3 concurrent queries. Considering that I still need to save the data to the database, it gives pretty high performance even without proxy.

Each parser function either saves the results in the database or passes the data to the next function. I don’t have to worry about additional data synchronization, and I get a high competitive execution value, which generates a low memory and CPU load.

async def category_loading(self, page):
  logger.info(f"Loading category page {page}")
  url = CATEGORY_URL.format(page=page)
  response_content = await self.downloader.get(url)
  json_data = json.loads(response_content)
  topics, more_topic = self.category_parser(json_data)
  if more_topic:
      next_page = int(page) + 1
      asyncio.create_task(self.category_loading(next_page)) # I call the same function with the new parameter
  for topic in topics:
      asyncio.create_task(self.topic_loader(topic)) # I create tasks for parsing all the found topics. That's up to 30 new requests

A very interesting point worth paying attention to is the stop method.

async def stop(self):
  while len(asyncio.all_tasks()) > 1: # The script will close when one task remains. The current function
      logger.warning(f"Current tasks pool {len(asyncio.all_tasks())}")
      await asyncio.sleep(10)
  logger.warning(f"Current tasks pool {len(asyncio.all_tasks())}")
  await self.downloader.stop()

Since I use create_task everywhere, I don’t wait for tasks to complete. If I don’t check to see whether the tasks are done working in the event loop, my script will finish almost instantly. And that’s not the behavior I want to achieve.

The script is completed when there is only 1 task left because that task is the actual completion task. If I made my script terminate at 0, then my script would never stop

The function to start the scan has a very simple appearance as a result:

async def start_scan():
  try:
      mongo = MongoManage()
      crowler = Crowler(db_manager=mongo)
      await crowler.start(start_page=0)
      await crowler.stop()
  except Exception as e:
      logger.exception(e)
  finally:
      await mongo.close()

Results

So we have 7,809 topics and 25,666 posts in them. In doing so, we have created a high-performance web parser. This statement is pretty trivial, but you have to follow all 3 steps to create scanners.

1. Scanning Strategy
2. Choice of technology
3. Сoding

Never start with point 3. You’ll make life difficult for yourself and the site owners.