Fetching Instagram Data Concurrently in Python: Async vs Threads at Scale

When I started working with Instagram data, making one API request at a time was fine for small tests.

It becomes a different problem when I need to fetch results or collect across hundreds or thousands of queries.

The bottleneck isn't usually Python itself. It's waiting for HTTP responses.

Instead of doing this:

request → wait → request → wait → request → wait

I can run multiple independent requests concurrently:

request ────────┐
request ────────┤
request ────────┤→ responses
request ────────┤
request ────────┘

For this example, I'm using HikerAPI as the Instagram REST API. It uses an x-access-key header and provides pay-per-request access, with 100 free requests for testing.

The basic Python request

I start with a normal synchronous request using requests:

import requests

headers = {"x-access-key": "YOUR_KEY"}

resp = requests.get(
    "https://api.hikerapi.com/v2/fbsearch/topsearch",
    params={"query": "travel photographer"},
    headers=headers
)

print(resp.json())

This works perfectly well when I only have a few requests.

The problem appears when I have many independent searches.

For example:

queries = [
    "travel photographer",
    "food blogger",
    "fashion creator",
    "fitness influencer",
    "photography",
    "travel blogger",
]

Calling the API sequentially means Python waits for every response before starting the next request.

for query in queries:
    resp = requests.get(
        "https://api.hikerapi.com/v2/fbsearch/topsearch",
        params={"query": query},
        headers=headers
    )

    print(resp.json())

For a larger workload, I prefer concurrency.

Option 1: Threads with requests

Because HTTP requests spend most of their time waiting on network I/O, Python threads are a straightforward solution.

I can use ThreadPoolExecutor from the standard library:

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

API_KEY = "YOUR_KEY"

headers = {
    "x-access-key": API_KEY
}

URL = "https://api.hikerapi.com/v2/fbsearch/topsearch"


def fetch_search(query):
    response = requests.get(
        URL,
        params={"query": query},
        headers=headers,
        timeout=30
    )

    response.raise_for_status()

    return {
        "query": query,
        "data": response.json()
    }


queries = [
    "travel photographer",
    "food blogger",
    "fashion creator",
    "fitness influencer",
    "photography",
    "travel blogger",
]


with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [
        executor.submit(fetch_search, query)
        for query in queries
    ]

    for future in as_completed(futures):
        result = future.result()
        print(result["query"])
        print(result["data"])

The important part here is:

ThreadPoolExecutor(max_workers=5)

Instead of waiting for one request to finish before starting another, I can have several requests in flight simultaneously.

I don't want to blindly increase max_workers, though.

More threads don't automatically mean more throughput.

My concurrency should be based on the API's rate limits, network latency, workload size, and how much pressure I want to put on the service.

Making the threaded version reusable

For an actual data collection script, I usually separate the API request from the concurrency logic.

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

API_KEY = "YOUR_KEY"
BASE_URL = "https://api.hikerapi.com"

HEADERS = {
    "x-access-key": API_KEY
}


def api_get(endpoint, params):
    response = requests.get(
        f"{BASE_URL}{endpoint}",
        params=params,
        headers=HEADERS,
        timeout=30
    )

    response.raise_for_status()
    return response.json()


def fetch_query(query):
    return api_get(
        "/v2/fbsearch/topsearch",
        {"query": query}
    )


def fetch_concurrently(queries, workers=5):
    results = []

    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_map = {
            executor.submit(fetch_query, query): query
            for query in queries
        }

        for future in as_completed(future_map):
            query = future_map[future]

            try:
                data = future.result()

                results.append({
                    "query": query,
                    "data": data
                })

            except requests.RequestException as exc:
                print(f"Request failed for {query}: {exc}")

    return results


queries = [
    "travel photographer",
    "food blogger",
    "fashion creator",
    "fitness influencer",
    "travel blogger",
]

results = fetch_concurrently(queries, workers=5)

print(f"Collected {len(results)} results")

This structure makes it easier to swap endpoints later.

For example, I can use the same concurrency layer for .

Option 2: Async requests with httpx

The second approach I use is asynchronous HTTP with httpx.

First, I install it:

pip install httpx

Then I can create asynchronous requests:

import asyncio
import httpx

API_KEY = "YOUR_KEY"

HEADERS = {
    "x-access-key": API_KEY
}

URL = "https://api.hikerapi.com/v2/fbsearch/topsearch"


async def fetch_search(client, query):
    response = await client.get(
        URL,
        params={"query": query},
        headers=HEADERS,
        timeout=30
    )

    response.raise_for_status()

    return {
        "query": query,
        "data": response.json()
    }


async def main():
    queries = [
        "travel photographer",
        "food blogger",
        "fashion creator",
        "fitness influencer",
        "photography",
        "travel blogger",
    ]

    async with httpx.AsyncClient() as client:
        tasks = [
            fetch_search(client, query)
            for query in queries
        ]

        results = await asyncio.gather(*tasks)

    for result in results:
        print(result["query"])
        print(result["data"])


asyncio.run(main())

This is useful when the application is already async, for example when I'm building an async web service, crawler, or data pipeline.

Limiting async concurrency

There's an important difference between concurrency and sending as many requests as possible.

I don't want to create thousands of tasks and let them all hit the API at once.

Instead, I can use an asyncio.Semaphore:

import asyncio
import httpx

API_KEY = "YOUR_KEY"

HEADERS = {
    "x-access-key": API_KEY
}

URL = "https://api.hikerapi.com/v2/fbsearch/topsearch"

semaphore = asyncio.Semaphore(5)


async def fetch_search(client, query):
    async with semaphore:
        response = await client.get(
            URL,
            params={"query": query},
            headers=HEADERS,
            timeout=30
        )

        response.raise_for_status()

        return {
            "query": query,
            "data": response.json()
        }


async def main():
    queries = [
        "travel photographer",
        "food blogger",
        "fashion creator",
        "fitness influencer",
        "travel blogger",
        "street photographer",
        "tech creator",
        "lifestyle blogger",
    ]

    async with httpx.AsyncClient() as client:
        tasks = [
            fetch_search(client, query)
            for query in queries
        ]

        results = await asyncio.gather(
            *tasks,
            return_exceptions=True
        )

    for result in results:
        if isinstance(result, Exception):
            print("Request failed:", result)
            continue

        print(result["query"])


asyncio.run(main())

Now I can create many tasks while only allowing five HTTP requests to execute concurrently.

That's much safer for larger workloads.

Handling rate limits

Concurrency doesn't remove rate limits.

It can actually make rate-limit problems appear much faster.

HikerAPI documents a default limit of 15 requests per second, with higher limits available depending on the account. So I treat the API limit as a constraint when deciding my concurrency level.

For example, if I'm allowed to make 15 requests per second, I don't simply launch 100 simultaneous requests and hope for the best.

I can add retry handling for temporary failures and back off before trying again.

import asyncio
import httpx


async def fetch_with_retry(
    client,
    query,
    max_retries=3
):
    for attempt in range(max_retries):
        try:
            response = await client.get(
                "https://api.hikerapi.com/v2/fbsearch/topsearch",
                params={"query": query},
                headers={"x-access-key": "YOUR_KEY"},
                timeout=30
            )

            if response.status_code == 429:
                wait_time = 2 ** attempt

                print(
                    f"Rate limited. "
                    f"Retrying in {wait_time}s..."
                )

                await asyncio.sleep(wait_time)
                continue

            response.raise_for_status()

            return response.json()

        except httpx.TimeoutException:
            if attempt == max_retries - 1:
                raise

            await asyncio.sleep(2 ** attempt)

    raise RuntimeError(
        f"Failed after {max_retries} retries: {query}"
    )

The key idea is that I don't treat a 429 as a normal successful response.

I slow down, wait, and retry.

For production workloads, I'd also consider respecting a server-provided Retry-After value when it is available rather than always using a fixed backoff.

Async vs threads: which one should I use?

For simple scripts, I usually find threads easier to understand.

If I'm already using requests, this:

with ThreadPoolExecutor(max_workers=5) as executor:
    ...

is a relatively small change from sequential code.

Async becomes more attractive when my application already uses asyncio or I need to coordinate many asynchronous operations.

Here's how I think about it:

Approach	Good for
requests sequential	Small workloads
ThreadPoolExecutor	Existing synchronous Python code
httpx.AsyncClient	Async applications and high I/O concurrency
Semaphore + async	Controlled high-volume workloads

The biggest improvement isn't necessarily choosing threads over async.

It's avoiding unnecessary sequential waiting.

A production-style async pattern

For larger jobs, I like combining a few ideas:

• Reuse an HTTP client
• Limit concurrency
• Handle timeouts
• Handle rate limits
• Capture failures
• Keep the input associated with the response

Here's a compact version:

import asyncio
import httpx

API_KEY = "YOUR_KEY"

BASE_URL = "https://api.hikerapi.com"

HEADERS = {
    "x-access-key": API_KEY
}

CONCURRENCY = 5


async def fetch(
    client,
    semaphore,
    query
):
    async with semaphore:

        for attempt in range(3):
            try:
                response = await client.get(
                    f"{BASE_URL}/v2/fbsearch/topsearch",
                    params={"query": query},
                    headers=HEADERS,
                    timeout=30
                )

                if response.status_code == 429:
                    await asyncio.sleep(2 ** attempt)
                    continue

                response.raise_for_status()

                return {
                    "query": query,
                    "data": response.json(),
                    "error": None
                }

            except Exception as exc:
                if attempt == 2:
                    return {
                        "query": query,
                        "data": None,
                        "error": str(exc)
                    }

                await asyncio.sleep(2 ** attempt)


async def main(queries):
    semaphore = asyncio.Semaphore(CONCURRENCY)

    async with httpx.AsyncClient() as client:
        tasks = [
            fetch(client, semaphore, query)
            for query in queries
        ]

        return await asyncio.gather(*tasks)


queries = [
    "travel photographer",
    "food blogger",
    "fashion creator",
    "fitness influencer",
    "travel blogger",
    "street photographer",
    "tech creator",
    "lifestyle blogger",
]

results = asyncio.run(main(queries))

successful = [
    result for result in results
    if result["error"] is None
]

failed = [
    result for result in results
    if result["error"] is not None
]

print(f"Successful: {len(successful)}")
print(f"Failed: {len(failed)}")

This is much closer to the pattern I'd use for a real data collection job.

What happens when the dataset gets bigger?

Suppose I'm processing:

<my volume> queries

I don't necessarily want to put every query into memory and create one task for each immediately.

For very large workloads, I can process the input in batches.

async def process_in_batches(queries, batch_size=100):
    all_results = []

    for start in range(0, len(queries), batch_size):
        batch = queries[start:start + batch_size]

        results = await main(batch)

        all_results.extend(results)

        print(
            f"Processed "
            f"{min(start + batch_size, len(queries))}"
            f"/{len(queries)}"
        )

    return all_results

This gives me another control layer:

Total workload
      ↓
Batch size
      ↓
Concurrency limit
      ↓
API requests

For example, I might experiment with:

batch_size = 100
CONCURRENCY = 5

and adjust these values based on actual response times and the API limits applicable to my account.

Don't confuse concurrency with scraping everything at once

One mistake I see with concurrent HTTP clients is assuming:

More workers = more speed.

That's only true up to a point.

If I increase concurrency from 5 to 50, I may run into:

• API rate limits
• more 429 responses
• connection overhead
• more retries
• increased memory usage
• unnecessary load
• harder-to-debug failures

The goal isn't maximum concurrency.

The goal is controlled throughput.

I want enough concurrent requests to keep the network busy without constantly hitting the API's limits.

Measuring the difference

Before changing my implementation, I can measure how long the sequential version takes.

import time
import requests

queries = [
    "travel photographer",
    "food blogger",
    "fashion creator",
    "fitness influencer",
    "travel blogger",
]

start = time.perf_counter()

for query in queries:
    requests.get(
        "https://api.hikerapi.com/v2/fbsearch/topsearch",
        params={"query": query},
        headers={"x-access-key": "YOUR_KEY"},
        timeout=30
    )

elapsed = time.perf_counter() - start

print(f"Elapsed: {elapsed:.2f}s")

Then I can compare it with the concurrent implementation.

The exact improvement depends on network latency, API response time, concurrency, rate limits, and the size of the workload, so I prefer measuring my actual workload rather than assuming a specific speedup.

My practical approach

For a small script, I'd keep it simple:

requests

For an existing synchronous application where I need several requests at the same time:

ThreadPoolExecutor

For an async application or a larger I/O-heavy pipeline:

httpx.AsyncClient
+
asyncio.Semaphore
+
retry/backoff

That gives me a good balance between throughput and control.

The API itself remains simple: send the request with the x-access-key header, receive JSON, and let Python handle multiple independent requests concurrently.

Final takeaway

When I'm fetching Instagram data at scale, the main optimization isn't complicated Python.

It's recognizing that HTTP requests are I/O-bound and that I don't need to wait for every request to finish before starting the next one.

Starting from this:

resp = requests.get(...)

I can move to:

ThreadPoolExecutor

or:

httpx.AsyncClient

and then add the controls that matter for production:

Concurrency
    ↓
Rate-limit awareness
    ↓
Retries + backoff
    ↓
Timeouts
    ↓
Batching
    ↓
Error tracking

For my own workload, I'd benchmark both approaches against and , then choose the implementation that gives me the best combination of throughput, reliability, and code simplicity.

python #api #async