How to filter and aggregate data from a REST API in Python: step by step

This tutorial shows how to filter and aggregate data received from a REST API in Python to obtain summarized, analyzable reports. Knowing how to filter and aggregate directly on the client side is useful when the API does not provide analytical endpoints or when you want to validate data before loading it into an ETL system.

Prerequisites

• Python 3.8+ installed.
• Libraries: requests and pandas (pip install requests pandas).
• A REST API that returns JSON with similar records (e.g., a listing of transactions or events).
• Text editor/IDE and terminal.

Step 1: understand the API format and choose filters

Before writing code, check the API documentation: query parameters, returned fields and pagination. Decide the filters you will apply (dates, status, type) and the keys to aggregate by (e.g., category, day, customer).

Step 2: fetch API pages with requests

Implement the API call with pagination. Minimal example: request per page until there are no more results. Handle HTTP errors and timeouts.

import requests

def fetch_all(url, params=None, headers=None, page_param='page'):
    params = params.copy() if params else {}
    page = 1
    all_items = []
    while True:
        params[page_param] = page
        resp = requests.get(url, params=params, headers=headers, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        items = data.get('results') or data.get('items') or data
        if not items:
            break
        all_items.extend(items)
        # simple stop condition: when fewer items than a typical page
        if isinstance(items, list) and len(items) < 100:
            break
        page += 1
    return all_items

Step 3: apply filters on the client

Filtering on the client is useful when the API does not accept all parameters. Apply filters in a clear and reproducible way. Example: filter by date and status.

from datetime import datetime

def filter_items(items, start_date=None, end_date=None, status=None):
    def in_range(d):
        if not d:
            return False
        dt = datetime.fromisoformat(d)
        if start_date and dt < start_date:
            return False
        if end_date and dt > end_date:
            return False
        return True

    filtered = []
    for it in items:
        if status and it.get('status') != status:
            continue
        if start_date or end_date:
            if not in_range(it.get('created_at')):
                continue
        filtered.append(it)
    return filtered

Step 4: transform JSON to a pandas DataFrame

Converting to a DataFrame makes aggregation and cleaning easier. Normalize nested lists or optional fields before aggregating.

import pandas as pd

def to_dataframe(items):
    df = pd.json_normalize(items)
    # convert date columns
    if 'created_at' in df.columns:
        df['created_at'] = pd.to_datetime(df['created_at'])
    return df

Step 5: aggregate and compute metrics

Use groupby to summarize by key. Example: sum amounts by day and by category, count transactions and compute average.

def aggregate_metrics(df):
    # create day column
    df['day'] = df['created_at'].dt.date
    # example metrics by day and category
    agg = (df.groupby(['day', 'category'])
             .agg(total_amount=pd.NamedAgg(column='amount', aggfunc='sum'),
                  count=pd.NamedAgg(column='id', aggfunc='count'),
                  avg_amount=pd.NamedAgg(column='amount', aggfunc='mean'))
             .reset_index())
    return agg

Step 6: handle common errors and performance

Common errors: unexpected JSON, timeout, rate limits and missing fields. For large volumes, paginate using the API parameters, process in batches and avoid loading everything into memory. Use chunks and write intermediate results to CSV/Parquet files.

# simple example of writing by chunks
for i in range(0, len(all_items), 1000):
    chunk = all_items[i:i+1000]
    df_chunk = to_dataframe(chunk)
    df_chunk.to_parquet(f'data_chunk_{i//1000}.parquet')

Verify the result

Open the aggregated DataFrame (agg) and confirm sums, counts and dates. Check some raw rows to confirm that the filters worked. Quick verification examples:

print(agg.head())
print(df[['id','created_at','status','amount']].sample(5))
print('Total original:', len(all_items), 'After filters:', len(df))

Conclusion

Filtering and aggregating data from a REST API in Python enables creating reports even when the API lacks analytical endpoints. Next steps: add authentication (Bearer/OAuth), use tools like Dask for large volumes or automate as part of an ETL pipeline. Tip: try storing outputs in Parquet for faster queries — what aggregation would make sense for your team?