reqstorm

Send thousands of HTTP requests without writing the plumbing.

reqstorm sends large numbers of HTTP requests concurrently and gives you one result per request. Rate limits, retries, progress, error reports and output to files or databases are built in.

📖 Documentation: reqstorm.github.io

reqstorm is the new name of reqt. Coming from reqt? See Coming from reqt.

import reqstorm

urls = [
    f"https://api.example.com/items/{i}"
    for i in range(7000)
]

print(reqstorm.estimate(7000, rate_limit="100/min"))
# 7000 requests: about 1h 10m
# (limited by rate_limit)

results = reqstorm.fetch_all_sync(
    urls,
    rate_limit="100/min",
    retries=2,
    progress=True,
)

print(results.summary())
# {'total': 7000, 'ok': 6987, 'failed': 13,
#  'failures': {'HTTP 404': 9, 'TimeoutError': 4}}

for error in results.errors():
    print(error["url"], error["error"] or error["status"])

Contents

• Why reqstorm
• Installation
• Quick start
• Results and reports
• Rate limits and concurrency
• Timeouts and retries
• Writing results to a file
• Writing results to a database
• Structured data: JSON to typed columns
• Requests, headers and bodies
• Streaming results
• TLS and sessions
• When to use something else
• Coming from reqt
• Development
• License

Why reqstorm

Sending one request is easy. Sending 7000 to an API that allows 100 a minute, without losing results when a few of them fail, is where the work is:

• Pacing: stay under the API's rate limit, per host, including retries.
• Isolating failures: one timeout or dropped connection must not stop the other 6999 requests.
• Retrying the right failures: retry 503s and timeouts, never 404s, and back off when the server says so.
• Keeping the results: write them somewhere as they arrive, and resume where you stopped if the run is interrupted.
• Knowing what happened: which requests failed, why, and after how many attempts.

reqstorm does all of that for you, with one call.

Installation

$ python -m pip install reqstorm

reqstorm supports Python 3.9 to 3.13 on Linux, macOS and Windows. Its only dependency is aiohttp. For PostgreSQL or MySQL output, install the driver you already use (psycopg, psycopg2, pymysql, mysqlclient or mysql-connector-python).

Quick start

Every function has a blocking version ending in _sync for scripts and notebooks, and an async version for code that already runs an event loop. Both take the same options.

In a script or a Jupyter notebook:

import reqstorm

urls = [
    f"https://httpbin.org/get?page={page}"
    for page in range(1, 101)
]

results = reqstorm.fetch_all_sync(
    urls,
    concurrency=20,
    timeout=10,
    retries=2,
)

for result in results:
    if result.ok:
        print(result.url, result.status)
    else:
        print(result.url, "failed:",
              result.error or result.status)

In async code:

import asyncio
import reqstorm

async def main():
    results = await reqstorm.fetch_all(
        urls, concurrency=20, timeout=10
    )
    print(results.summary())

asyncio.run(main())

Results and reports

fetch_all returns a Results list with one Result per request, in the order you gave them. A failed request never raises an exception; it is recorded on its result.

results.succeeded    # successful results
results.failed       # failed results
results.summary()    # counts, failures grouped by reason
results.errors()     # failed requests as plain dicts
results.to_dicts()   # every result as a dict

errors() returns plain data, ready for a log file or a DataFrame:

{
  "index": 12,
  "method": "GET",
  "url": "https://api.example.com/items/12",
  "status": null,
  "ok": false,
  "error": "TimeoutError",
  "attempts": 3,
  "elapsed": 20.415,
  "history": [
    {"attempt": 1, "status": null,
     "error": "TimeoutError", "elapsed": 10.001},
    {"attempt": 2, "status": 503,
     "error": null, "elapsed": 0.412},
    {"attempt": 3, "status": null,
     "error": "TimeoutError", "elapsed": 10.002}
  ]
}

Prefer exceptions? result.raise_for_error() raises the request's error, or reqstorm.HTTPStatusError for a status of 400 or above.

Rate limits and concurrency

results = reqstorm.fetch_all_sync(
    urls,
    rate_limit="100/min",     # per host
    concurrency=50,           # in flight, in total
    concurrency_per_host=10,  # in flight, per host
)

The limit applies to each host (host:port) separately and counts retries too, so requests to different APIs never slow each other down. Requests to one host are spaced evenly.

Plan before you send. estimate predicts the duration and names the bottleneck, without sending anything:

>>> print(reqstorm.estimate(7000, rate_limit="100/min"))
7000 requests: about 1h 10m (limited by rate_limit)
>>> print(reqstorm.estimate(
...     7000, rate_limit="100/min", hosts=7))
7000 requests: about 10m 00s (limited by rate_limit)

Watch it run. progress=True prints progress to stderr:

reqstorm: 3500/7000 (50%)  ok 3493  failed 7
          1.7 req/s  ETA 35m 00s

Timeouts and retries

results = reqstorm.fetch_all_sync(
    urls,
    timeout=10,            # seconds per attempt
    retries=3,             # retry right away...
    backoff=0.5,           # ...0.5 s, 1 s, 2 s apart
    retry_rounds=2,        # then resend what still
    retry_round_delay=30,  # failed, 30 s later
)

• timeout is the time allowed for one attempt, including reading the body (default 30 s; None disables it). A slow server only fails its own requests.
• retries retries a request right away. The delay starts at backoff and doubles; a Retry-After header (up to 60 s) takes precedence.
• retry_rounds holds back requests that still failed for a retryable reason and sends them again after the rest of the batch, for temporary outages. Each request still produces exactly one result, with all its attempts in history.

Writing results to a file

For large batches, write results as they arrive instead of keeping them in memory:

summary = reqstorm.fetch_to_file_sync(
    urls,
    "results.jsonl",
    rate_limit="10/s",
    progress=True,
    resume=True,
)
print(summary.ok, summary.failed, summary.skipped)
print(summary.errors[:5])  # failed records

• Format follows the file name: .jsonl, .csv, or .db / .sqlite / .sqlite3 for SQLite. Pass format= to override it.
• Order: records are written as requests complete; ordered=True writes them in input order instead (finished records wait in memory for the ones before them).
• Resume: with resume=True, requests already recorded as successful are skipped and the rest are sent and appended. A line cut short by an interruption is ignored. Without resume, a JSONL or CSV file is overwritten.
• Fields: index, method, url, status, ok, error, attempts, elapsed, final_url, history and body.
• Body: body="text" (default), "base64" for binary responses, or "none". include_headers=True adds the response headers.

Writing results to a database

fetch_to_db inserts one row per request through a connection you already have:

import psycopg
import reqstorm

with psycopg.connect("dbname=crawl") as connection:
    summary = reqstorm.fetch_to_db_sync(
        urls,
        connection,
        table="api_results",
        resume=True,
    )

Supported drivers: sqlite3, psycopg and psycopg2 (PostgreSQL), pymysql, MySQLdb and mysql.connector (MySQL). For SQLite a file name is enough: fetch_to_file_sync(urls, "results.db").

reqstorm creates the table if it does not exist. The fields you filter on are real columns; the parts whose shape varies are JSON:

So the questions you ask afterwards are plain SQL:

SELECT url, error FROM api_results WHERE NOT ok;
SELECT status, count(*) FROM api_results
GROUP BY status;

Rows are inserted in batches on a background thread, so a remote database does not slow the requests down. Existing rows are never deleted. body="bytes" stores the raw body in a binary column.

Structured data: JSON to typed columns

Instead of storing raw responses, give a schema and each JSON field goes to its own typed column. Every value is checked first, so a number column never gets a string or a NaN.

from reqstorm import Field

schema = {
    "id": Field("id", int, required=True, key=True),
    "name": Field("name", str, required=True),
    "price": Field("pricing.amount", float),
    "tags": Field("tags", "json"),
    "updated": Field("updated_at", "datetime"),
    "page": Field("$.meta.page", int),
}

summary = reqstorm.fetch_to_db_sync(
    urls,
    connection,
    table="products",
    schema=schema,
    explode="items",  # one row per array element
    rejects_table="products_rejects",
)
print(summary.rows, summary.rejected)

• Types: int, float, str, bool, "datetime" and "json" become BIGINT, DOUBLE PRECISION, TEXT, BOOLEAN, TIMESTAMPTZ and JSONB in PostgreSQL, and their equivalents in MySQL and SQLite.
• Strict checks: 10.5 is rejected for an int, "42" for a number, and NaN or Infinity always. With coerce=True, compatible values such as "12.5" are converted.
• Nested JSON: dotted paths read nested values ("pricing.amount", "tags.0"). Type "json" keeps a part whole. explode makes each array element a row, and "$." paths read from the response root.
• Rejected records are never written, not even partly. They are listed in summary.errors with their reasons and, with rejects_table, stored with the original record.
• No duplicates: key=True fields form the primary key. Running the batch again updates existing rows.
• Resume works: each row records the request it came from (source_url).

The same schema works for .db, .jsonl and .csv files, and reqstorm.extract(results, schema) returns the rows as Python lists. reqstorm.infer_schema(samples) drafts a schema from a few responses for you to review.

More in the structured data guide.

Requests, headers and bodies

Options given to fetch_all apply to every request; reqstorm.Request varies them per request:

results = reqstorm.fetch_all_sync(
    [
        "https://api.example.com/items/1",
        reqstorm.Request(
            "https://api.example.com/items",
            method="POST",
            json={"name": "new"},
        ),
    ],
    headers={"Authorization": "Bearer ..."},
)

• fetch_all_sync(urls, "POST", json=...) sends every plain URL as a POST.
• params, json and data work the same way.
• A Request's own headers are merged on top of the shared ones; its other fields replace the shared value.

Streaming results

stream_sync (or stream in async code) yields each result as soon as it completes. URLs can come from a generator, which is read lazily, so millions of URLs never need to fit in memory:

def read_urls():
    with open("urls.txt") as file:
        for line in file:
            yield line.strip()

for result in reqstorm.stream_sync(read_urls()):
    save(result.url, result.status, result.body)

Leaving the loop early (break) cancels the requests still in flight. fetch_all also accepts callback=, called with each result as it completes.

TLS and sessions

TLS certificates are verified by default. To trust a private certificate authority, pass an ssl.SSLContext:

import ssl

context = ssl.create_default_context(
    cafile="internal-ca.pem"
)
results = reqstorm.fetch_all_sync(urls, ssl=context)

verify_ssl=False turns verification off; only use it for hosts you control.

In async code, session= takes an existing aiohttp.ClientSession to share cookies, connection pools or proxy settings. reqstorm does not close it.

When to use something else

reqstorm is built for batches. For other jobs, these are better fits:

reqstorm uses aiohttp underneath and adds the batch work: pacing, retries, failure isolation, reports and output.

Coming from reqt

reqstorm 2.0.1 is the next version of reqt, renamed because another project already uses the reqt name.

$ python -m pip install reqstorm

Then replace import reqt with import reqstorm. The reqt package on PyPI now only installs reqstorm and re-exports it with a deprecation warning, so existing code keeps working while you switch.

The reqt 1.x style, which called a function with each raw response, still works with a DeprecationWarning and will be removed in reqstorm 3.0:

async def handle(response):  # reqt 1.x style
    print(response.status)

await reqstorm.fetch_all(urls=urls, method=handle)

# The same today:
for result in reqstorm.fetch_all_sync(urls):
    print(result.status)

Two things changed even in the 1.x style:

• TLS certificates are now verified. reqt 1.x skipped verification. Use verify_ssl=False only for hosts you control.
• Failures no longer stop the batch. Every failed request is logged to the reqstorm logger.

Development

$ git clone https://github.com/melihcolpan/reqstorm
$ cd reqstorm
$ python -m pip install -e ".[test,lint]"
$ pytest
$ ruff check . && ruff format --check . && mypy reqstorm

• The tests run against a local server and a throwaway certificate authority; they need no network access.
• The PostgreSQL and MySQL tests run when REQSTORM_TEST_POSTGRES (a libpq connection string) and REQSTORM_TEST_MYSQL (host:port:user:password:database) are set.
• The documentation is in docs/ and builds with pip install -r docs/requirements.txt && mkdocs serve.

CI runs the tests on Python 3.9 to 3.13 (Linux, plus macOS and Windows), against real PostgreSQL and MySQL, and builds the package and the documentation. A GitHub release tagged vX.Y.Z publishes to PyPI through trusted publishing.

Bug reports and pull requests are welcome on GitHub.

License

MIT, see LICENSE.