Make Direct Async Requests to SerpApi with Python
Tags: web, webscraping, api, tutorial
Published: 2023-05-24 12:19:59

## Intro

In the previous [SerpApi Async Requests with Pagination using Python](https://serpapi.com/blog/serpapi-async-requests-with-pagination-using-python/) blog post we've covered how to make Async requests with [SerpApi's pagination](https://serpapi.com/pagination), how to use [Search Archive API](https://serpapi.com/search-archive-api) and `Queue`.

In this blog post we'll cover on how to make direct requests to `serpapi.com/search.json` without using SerpApi's [`google-search-results`](https://github.com/serpapi/google-search-results-python) Python client.

This way, when making a direct request SerpApi, we can get a slightly faster response time in comparison to Python's client [batch `async` search](https://github.com/serpapi/google-search-results-python#batch-asynchronous-searches) feature which uses `Queue`.

For example, 50 requests in less than 10 seconds. Depends on the internet speed.

In the following blog post, we'll cover how to add pagination to the shown code below.

______

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

**Test includes**: 50 async search queries.


## Code

You can [check the code example in the online IDE](https://replit.com/@DimitryZub1/SerpApi-Direct-Async-Multiple-Requests?v=1):

```python
import aiohttp
import asyncio
import json
import time

async def fetch_results(session, query):
    params = {
        'api_key': '...',      # your serpapi api key: https://serpapi.com/manage-api-key
        'engine': 'youtube',   # search engine to parse data from
        'device': 'desktop',   # from which device to parse data
        'search_query': query, # search query
        'no_cache': 'true'     # https://serpapi.com/search-api#api-parameters-serpapi-parameters-no-cache
    }
    
    async with session.get('https://serpapi.com/search.json', params=params) as response:
        results = await response.json()

    data = []

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

    return data

async def main():
    # 50 queries
    queries = [
        'burly',
        'creator',
        'doubtful',
        'chance',
        'capable',
        'window',
        'dynamic',
        'train',
        'worry',
        'useless',
        'steady',
        'thoughtful',
        'matter',
        'rotten',
        'overflow',
        'object',
        'far-flung',
        'gabby',
        'tiresome',
        'scatter',
        'exclusive',
        'wealth',
        'yummy',
        'play',
        'saw',
        'spiteful',
        'perform',
        'busy',
        'hypnotic',
        'sniff',
        'early',
        'mindless',
        'airplane',
        'distribution',
        'ahead',
        'good',
        'squeeze',
        'ship',
        'excuse',
        'chubby',
        'smiling',
        'wide',
        'structure',
        'wrap',
        'point',
        'file',
        'sack',
        'slope',
        'therapeutic',
        'disturbed'
    ]

    data = []

    async with aiohttp.ClientSession() as session:
        tasks = []
        for query in queries:
            task = asyncio.ensure_future(fetch_results(session, query))
            tasks.append(task)

        start_time = time.time()
        results = await asyncio.gather(*tasks)
        end_time = time.time()

        data = [item for sublist in results for item in sublist]

    print(json.dumps(data, indent=2, ensure_ascii=False))
    print(f'Script execution time: {end_time - start_time} seconds') # ~7.192448616027832 seconds

asyncio.run(main())
```

### Code Explanation

Import libraries:

```python
import aiohttp # to make a request
import asyncio
import json    # for printing data
import time    # to measure execution time
```

In the `fetch_results()` function we:

1. creating search `params` that will be passed to SerpApi while making request.
2. making an `async` session request, passing params and awaiting for each `response` and storing it to `results` variable.
3. checking for `'error'` in the `results` and iterating over `'video_results'`, and storing extracted data to the `data` `list`.
4. returning `list` with videos data.

```python
async def fetch_results(session, query):
    params = {
        'api_key': '...',      # your serpapi api key: https://serpapi.com/manage-api-key
        'engine': 'youtube',   # search engine to parse data from
        'device': 'desktop',   # from which device to parse data
        'search_query': query, # search query
        'no_cache': 'true'     # https://serpapi.com/search-api#api-parameters-serpapi-parameters-no-cache
    }
    
    async with session.get('https://serpapi.com/search.json', params=params) as response:
        results = await response.json()

    data = []

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

    return data
```



In the second `main()` function we:

1. create a `list` of `queries`. Could be also txt/csv/json.
2. open a `aiohttp.ClientSession()`.
3. iterate over queries and create `asyncio` tasks.
4. proceed all of the tasks with `asyncio.gather(*tasks)`.
5. flatten `list` with data and store it to the `data` variable.
6. print the data.

```python
async def main():
    queries = [
        'burly',
        'creator',
        'doubtful',
        # ...
    ]

    data = []

    async with aiohttp.ClientSession() as session:
        tasks = []
        for query in queries:
            task = asyncio.ensure_future(fetch_results(session, query))
            tasks.append(task)

        start_time = time.time()
        results = await asyncio.gather(*tasks)
        end_time = time.time()

        data = [item for sublist in results for item in sublist]

    print(json.dumps(data, indent=2, ensure_ascii=False))
    print(f'Script execution time: {end_time - start_time} seconds') # ~7.192448616027832 seconds

asyncio.run(main())
```

## Conclusion

As you saw (and possibly tried) this results in quite a fast response times. Additionally, we can add pagination to it, which will be covered in the next blog post. 

<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>