DEV Community: Muhammad Sheraz

How I Fetch Instagram Data Concurrently in Python with HikerAPI

Muhammad Sheraz — Thu, 03 Sep 2026 18:23:49 +0000

When building tools for a platform like StyleGen, I sometimes need to work with multiple independent API requests instead of processing everything one request at a time.

For example, if I'm working with and need to process Instagram-related queries, sequential HTTP requests can become unnecessarily slow.

The basic request is simple.

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

The interesting part starts when I have many queries.

Why I Use Concurrent Requests

Imagine I have these searches:

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

The simplest implementation is sequential:

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

    print(resp.json())

The problem is that Python waits for the first HTTP response before starting the second request.

For an API-driven application, that waiting time adds up quickly.

Instead, I can have several independent requests running concurrently.

There are two approaches I commonly consider in Python:

Threads
Async HTTP

Using Threads with requests

If I'm already using requests, ThreadPoolExecutor is probably the easiest way to introduce concurrency without rewriting the whole application.

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",
    "tech creator",
    "lifestyle 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 is:

ThreadPoolExecutor(max_workers=5)

Instead of waiting for every request individually, I can have up to five requests being processed concurrently.

This works well because HTTP requests are largely I/O-bound. While one request is waiting for the server, another thread can work on a different request.

Keeping the API layer separate

For a larger StyleGen-related project, I wouldn't put the HTTP request logic directly inside the business logic.

I prefer creating a small API function:

import requests

API_KEY = "YOUR_KEY"

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

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


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

Now my Instagram search function becomes very small:

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

And I can reuse the same pattern for other endpoints.

Async Python with httpx

If I'm building an asynchronous application, I prefer an async HTTP client such as httpx.

I install it with:

pip install httpx

Then I can make the same API call asynchronously:

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",
        "tech creator",
        "lifestyle 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"])


asyncio.run(main())

The important difference is that I can create multiple async tasks and let the event loop coordinate the network operations.

Controlling Async Concurrency

I don't want to create an unlimited number of simultaneous requests.

If I have hundreds or thousands of queries, I need a concurrency limit.

An asyncio.Semaphore works well for this:

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",
        "tech creator",
        "lifestyle blogger",
        "street photographer",
        "content creator",
    ]

    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 have many tasks waiting while only five requests are allowed to execute concurrently.

That distinction becomes important at scale.

Rate Limits Still Matter

Concurrency isn't a way around API rate limits.

In fact, concurrency can make rate-limit problems happen much faster.

For my implementation, I therefore treat the API's applicable rate limit as a hard constraint.

If I receive a 429 Too Many Requests, I don't immediately send another request.

I back off first.

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"Waiting {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"
    )

This gives me exponential backoff:

Attempt 1 → wait 1 second
Attempt 2 → wait 2 seconds
Attempt 3 → wait 4 seconds

In production, I'd also use a server-provided Retry-After value when one is available.

Processing Large Workloads in Batches

If I'm processing queries, I don't necessarily want to launch everything simultaneously.

I can divide the workload into 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 multiple layers of control:

Total queries
      ↓
Batch size
      ↓
Concurrency limit
      ↓
API requests

For example:

batch_size = 100
concurrency = 5

I can then tune those values based on the workload and the rate limits applicable to my API account.

Threads vs Async

I don't think one approach is universally better.

I choose based on the application.

Approach	When I'd use it
`requests`	Small number of requests
`ThreadPoolExecutor`	Existing synchronous Python application
`httpx.AsyncClient`	Async application
Async + semaphore	Larger I/O-heavy workloads

If I already have a synchronous codebase, threads are often the easiest upgrade.

If the rest of my application is asynchronous, I would generally stay async rather than mixing concurrency models unnecessarily.

A Practical Production Pattern

For an application that needs to fetch Instagram data repeatedly, I'd combine:

A reusable HTTP client
Controlled concurrency
Timeouts
Rate-limit handling
Retries
Exponential backoff
Error tracking
Batch processing

The resulting architecture looks something like this:

Input queries
      ↓
Batch processing
      ↓
Concurrency limiter
      ↓
HikerAPI
      ↓
JSON responses
      ↓
Validation / error handling
      ↓
StyleGen feature

That separation is useful because the API-fetching layer doesn't need to know exactly what I'm doing with the returned data.

It could feed , another internal process, or a data pipeline.

Don't Just Increase max_workers

One of the easiest mistakes is assuming that this:

ThreadPoolExecutor(max_workers=50)

is automatically better than:

ThreadPoolExecutor(max_workers=5)

It isn't.

Higher concurrency can increase:

429 responses
retries
connection pressure
memory usage
failure rates

The goal isn't to maximize the number of simultaneous requests.

The goal is to find the highest stable throughput allowed by the API and my application.

Measure Before Optimizing

I also like measuring the baseline first.

For example:

import time
import requests

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 that with the concurrent implementation.

I don't assume a particular speedup because the result depends on:

API response time
Network latency
Number of requests
Concurrency
Rate limits
Retry frequency
Payload size

I measure the workload I'm actually running.

What I'd Use for StyleGen

For a small feature on StyleGen, I'd keep the implementation simple.

If I'm making only a handful of API requests:

requests

is enough.

If I already have synchronous Python code and want several requests running concurrently:

ThreadPoolExecutor

is a practical choice.

For an asynchronous service or larger I/O-heavy workload:

httpx.AsyncClient
+
asyncio.Semaphore
+
retry/backoff

gives me more control.

The important part isn't simply making requests faster.

It's building the request layer so that it remains predictable when the number of requests grows.

Final Takeaway

Fetching Instagram data sequentially is easy to understand:

response = requests.get(...)

But when I start processing , waiting for every HTTP request individually becomes unnecessary.

With a relatively small amount of Python code, I can parallelize the work:

requests
   ↓
ThreadPoolExecutor

or:

httpx
   ↓
asyncio
   ↓
Semaphore

Then I add the production safeguards:

Concurrency control
       ↓
Rate-limit handling
       ↓
Retries
       ↓
Exponential backoff
       ↓
Timeouts
       ↓
Batch processing
       ↓
Error tracking

That's the pattern I'd use when integrating Instagram API-driven functionality into a platform like StyleGen: concurrent enough to be efficient, but controlled enough to be reliable.

python #api #async

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

Muhammad Sheraz — Thu, 03 Sep 2026 13:14:54 +0000

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.