Async Requests with SerpApi and Python
Tags: programming, python, webscraping
Published: 2022-11-03 05:43:31

## Intro

Async requests are necessary when you need to get a lot of data quickly or not in real time.

**This blog post is about**:

- How to make async request with SerpApi.
- Understand [`async` parameter](https://serpapi.com/search-api#api-parameters-serpapi-parameters-async).
- What is [`Queue`](https://github.com/serpapi/google-search-results-python#batch-asynchronous-searches) and how to use use.
- [Search Archive API](https://serpapi.com/search-archive-api) and how to retrieve data.

**The subject of test**: [YouTube Search Engine Results API](https://serpapi.com/youtube-search-api).

**The test includes**: 500 YouTube search requests, and extraction of the data.

## What is Async parameter

We have a [`async` parameter](https://serpapi.com/search-api#api-parameters-serpapi-parameters-async) that tells SerpApi not to wait for the search to be completed thus allows to send more requests faster. The search will be sent and processed on the SerpApi backend.

After all requests have been sent, the data will be extracted from the [Search Archive API](https://serpapi.com/search-archive-api) and checked if the search succeeded or not.

📌Note: This blog post does not cover multithreading.

## Time Comparison

Time was recorded using [`$ time python <file.py>`](https://stackoverflow.com/a/1557577/15164646): 

| Type | Sync requests  | Async requests (no pagination)  | % difference     |
|------|----------------|---------------------------------|------------------|
| real | 13m 43.311s    | 5m 3.663s                       | +36.88% increase |
| user | 0m 5.942s      | 0m 11.222s                      | -52.95% decrease |
| sys  | 0m 0.813s      | 0m 1.191s                       | -68.26% decrease |


![image](https://user-images.githubusercontent.com/78694043/199235416-17c68491-b107-4cac-9f56-543cf7bc331a.png)

## SerpApi Sync Requests

```python
from serpapi import YoutubeSearch
import os, re, json

# shortened for the sake of not making code very long
queries = [
    'burly',
    'silk',
    'monkey',
    'abortive',
    'hot'
]

data = []

for query in queries:
    params = {
        'api_key': os.getenv('API_KEY'),  # serpapi api key
        'engine': 'youtube',              # search engine
        'device': 'desktop',              # device type
        'search_query': query,            # search query
    }

    search = YoutubeSearch(params)       # where data extraction happens
    results = search.get_dict()          # JSON -> Python dict

    if 'error' in results:
        print(results['error'])
        break

    for result in results.get('video_results', []):
        data.append({
            'title': result.get('title'),
            'link': result.get('link'),
            'channel': result.get('channel').get('name'),
        })

print(json.dumps(data, indent=2, ensure_ascii=False))
```

### Sync Code Explanation

Import libraries:

```python
from serpapi import YoutubeSearch
import os, re, json
```

Create a [`list`](https://www.w3schools.com/python/python_lists.asp) of search queries you want to search:

```python
queries = [
    'burly',
    'silk',
    'monkey',
    'abortive',
    'hot'
]
```

Create a temporary `list` that will store extracted data:

```python
data = []
```

Add a `for` loop to iterate over all `queries`, create [SerpApi YouTube search parameters](https://serpapi.com/youtube-search-api#api-parameters), and pass them `YoutubeSearch` which will make a request to SerpApi:

```python
for query in queries:
    params = {
        'api_key': os.getenv('API_KEY'),  # serpapi api key
        'engine': 'youtube',              # search engine
        'device': 'desktop',              # device type
        'search_query': query,            # search query
    }

    search = YoutubeSearch(params)       # where data extraction happens
    results = search.get_dict()          # JSON -> Python dict
```

[Check for `'errors'`](https://github.com/serpapi/google-search-results-python#error-management), iterate over video results and extract needed data to the temporary `list`, and print it: 

```python
if 'error' in results:
        print(results['error'])
        break

for result in results.get('video_results', []):
    data.append({
        'title': result.get('title'),
        'link': result.get('link'),
        'channel': result.get('channel').get('name'),
    })

print(json.dumps(data, indent=2, ensure_ascii=False))
```

## SerpApi Async Batch Requests without Pagination

```python
from serpapi import YoutubeSearch
from urllib.parse import (parse_qsl, urlsplit)
from queue import Queue
import os, re, json

queries = [
    'burly',
    'silk',
    'monkey',
    'abortive',
    'hot'
]

search_queue = Queue()

for query in queries:
    params = {
        'api_key': os.getenv('API_KEY'),  # serpapi api key
        'engine': 'youtube',              # search engine
        'device': 'desktop',              # device type
        'search_query': query,            # search query
        'async': True,                    # async batch requests
    }

    search = YoutubeSearch(params)       # where data extraction happens
    results = search.get_dict()          # JSON -> Python dict
    
    if 'error' in results:
        print(results['error'])
        break

    print(f"add search to the queue with ID: {results['search_metadata']}")
    search_queue.put(results)
    

data = []

while not search_queue.empty():
    result = search_queue.get()
    search_id = result['search_metadata']['id']

    print(f'Get search from archive: {search_id}')
    search_archived = search.get_search_archive(search_id)
    
    print(f"Search ID: {search_id}, Status: {search_archived['search_metadata']['status']}")

    if re.search(r'Cached|Success', search_archived['search_metadata']['status']):
        for result in search_archived.get('video_results', []):
            data.append({
                'title': result.get('title'),
                'link': result.get('link'),
                'channel': result.get('channel').get('name'),
            })
    else:
        print(f'Requeue search: {search_id}')
        search_queue.put(result)
        
print(json.dumps(data, indent=2))
print('all searches completed')
```

### Async Batch Requests without Pagination Explanation

Same as before, import libraries (a few more):

```python
from serpapi import YoutubeSearch
from urllib.parse import (parse_qsl, urlsplit)
from queue import Queue
import os, re, json
```

Create a `list` of search queries you want to search:

```python
queries = [
    'burly',
    'silk',
    'monkey',
    'abortive',
    'hot'
]
```

[Create a `Queue`](https://docs.python.org/3/library/queue.html#module-queue) to store all the requests that have been sent to SerpApi:

```python
search_queue = Queue()
```

Iterate over all queries, create [SerpApi YouTube search parameters](https://serpapi.com/youtube-search-api#api-parameters) with `'async': True` parameter present. Check for errors and [`put()` `search` in the queue](https://docs.python.org/3/library/queue.html#queue.Queue.put):

```python
for query in queries:
    params = {
        'api_key': os.getenv('API_KEY'),  # serpapi api key
        'engine': 'youtube',              # search engine
        'device': 'desktop',              # device type
        'search_query': query,            # search query
        'async': True,                    # async batch requests
    }

    search = YoutubeSearch(params)       # where data extraction happens
    results = search.get_dict()          # JSON -> Python dict
    
    if 'error' in results:
        print(results['error'])
        break

    print(f"add search to the queue with ID: {results['search_metadata']}")
    search_queue.put(results)
```

Create a temporary `list` that will be used to store extracted data from the search archive API:

```python
data = []
```

[Iterate over all queue until it's `empty()`](https://docs.python.org/3/library/queue.html#queue.Queue.empty) and get the data from search archive by accessing search ID:

```python
while not search_queue.empty():
    result = search_queue.get()
    search_id = result['search_metadata']['id']

    print(f'Get search from archive: {search_id}')
    search_archived = search.get_search_archive(search_id)
    
    print(f"Search ID: {search_id}, Status: {search_archived['search_metadata']['status']}")
```


Check if the search is either cached or succeeded, if so, extract needed data, otherwise we need to requeue the result:

```python
if re.search(r'Cached|Success', search_archived['search_metadata']['status']):
        for result in search_archived.get('video_results', []):
            data.append({
                'title': result.get('title'),
                'link': result.get('link'),
                'channel': result.get('channel').get('name'),
            })
    else:
        print(f'Requeue search: {search_id}')
        search_queue.put(result)
        
print(json.dumps(data, indent=2))
print('all searches completed')
```

That's basically it 🙂

To sum-up:

1. Send all the requests and store them in the queue.
2. When all requests have been sent, grab them one by one from the queue until the queue is empty.
3. Extract search ID. The search ID will not be random, it will be the right one from the queue it was earlier stored.
4. Check if the search is succeeded and extract the data. If not, requeue it.


## What comes next

In the next blog post that will be specifically about Async requests using SerpApi, we'll cover the following:

- how to speed up async requests i.e. multithreading `Queue`.
- how to add pagination with `async` parameter.


<p style="text-align: center;">Join us on <a href="https://twitter.com/serp_api">Twitter</a> | <a href="https://www.youtube.com/channel/UCUgIHlYBOD3yA3yDIRhg_mg">YouTube</a> </p>