Muhammad Sheraz

Posted on Sep 3

How I Fetch Instagram Data Concurrently in Python with HikerAPI

#python #api #programming

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

Top comments (1)

Marcus Kim • Sep 3

Capping both ThreadPoolExecutor and asyncio.Semaphore at five makes the central point concrete: throughput should be controlled, not simply maximized. The batch size of 100 and exponential backoff on 429s add useful boundaries, though I'd also honor Retry-After, add jitter, and retry selected 5xx responses so synchronized workers don't create another traffic spike. In production, I'd tune concurrency from observed latency and rate-limit frequency rather than keep it fixed, because the fastest stable setting can change as the provider or workload changes.