newsdataapi is the official Python SDK for the NewsData.io REST API. It wraps every endpoint (latest, archive, sources, crypto, market, count, crypto/count, market/count) with consistent retry, pagination, and error handling. It also covers the real-time WebSocket service end to end with NewsDataApiWebSocket: register, list, and delete queries, and stream the matching news as it is published (sync or asyncio).
pip install newsdataapiIf you use uv:
uv add newsdataapiSupports Python 3.10 through 3.14. The runtime dependencies are requests (REST) and websockets (real-time streaming).
from newsdataapi import NewsDataApiClient
with NewsDataApiClient("YOUR_API_KEY") as client:
response = client.latest_api(q="bitcoin", country="us", language="en")
for article in response["results"]:
print(article["title"], "-", article["link"])The context-manager form closes the underlying HTTP session cleanly when the block exits. If you prefer not to use with, create the client directly and call client.close() yourself:
from newsdataapi import NewsDataApiClient
client = NewsDataApiClient("YOUR_API_KEY")
try:
response = client.latest_api(q="bitcoin", country="us", language="en")
for article in response["results"]:
print(article["title"], "-", article["link"])
finally:
client.close()| Method | Endpoint | Notes |
|---|---|---|
latest_api() |
/latest |
Real-time news |
archive_api() |
/archive |
Historical news |
sources_api() |
/sources |
Available news sources |
crypto_api() |
/crypto |
Cryptocurrency news |
market_api() |
/market |
Market / financial news |
count_api(from_date, to_date) |
/count |
Aggregate counts |
crypto_count_api(from_date, to_date) |
/crypto/count |
Aggregate crypto counts |
market_count_api(from_date, to_date) |
/market/count |
Aggregate market counts |
All endpoint parameters are keyword-only (except the required from_date / to_date on the count endpoints). Most accept either a single string or a list[str]; lists are comma-joined for the API. The real-time WebSocket endpoints are covered by NewsDataApiWebSocket (see below).
See the NewsData.io documentation — or the OpenAPI 3.1 spec — for the full parameter reference.
# 1. Single request (the default).
response = client.latest_api(q="news")
# 2. Auto-merge — follow nextPage cursors and return one combined dict.
merged = client.latest_api(q="news", scroll=True, max_result=200)
# 3. Iterate one response per page (a generator).
for page in client.latest_api(q="news", paginate=True, max_pages=5):
process(page["results"])scroll and paginate are mutually exclusive. scroll=True truncates strictly to max_result; paginate=True stops at max_pages or when the API returns no nextPage.
Register a query first — the returned registration_id identifies it from then on:
from newsdataapi import NewsDataApiClient, NewsDataApiWebSocket
client = NewsDataApiClient("YOUR_API_KEY")
ws = NewsDataApiWebSocket(client)
response = ws.websocket_register(q="bitcoin", language="en")
registration_id = response["results"]["registration_id"]websocket_register accepts the familiar filter parameters (q, country, language, domain, …). Registering an identical query twice raises NewsdataAPIError with status_code=409 — the existing id is in e.response_body["results"]["registration_id"]. websocket_fetch() lists every registered query, and websocket_delete(registration_id) removes one.
Then stream — each yielded response has the familiar status / totalResults / results shape:
for response in ws.stream(registration_id):
for article in response["results"]:
print(article["title"], "-", article["link"])Use it as a context manager to close the connection promptly when you stop early (otherwise it closes when iteration ends):
with NewsDataApiWebSocket(client) as ws:
for response in ws.stream(registration_id):
print(response["totalResults"])
breakInside asyncio applications use stream_async() — the same class, same behavior, awaited iteration:
import asyncio
async def main():
async with NewsDataApiWebSocket(client) as ws:
async for response in ws.stream_async(registration_id):
for article in response["results"]:
print(article["title"], "-", article["link"])
asyncio.run(main())Transient drops (network errors, server restarts, abnormal closes) are reconnected automatically with a capped exponential backoff. Pass reconnect=False to stop on the first disconnect instead. A permanent rejection — bad API key, missing WebSocket entitlement, unknown registration_id, device limit reached, or exhausted quota — raises NewsdataWebSocketAuthError and is not retried:
from newsdataapi import NewsdataWebSocketAuthError, NewsdataWebSocketError
try:
for response in NewsDataApiWebSocket(client).stream(registration_id):
...
except NewsdataWebSocketAuthError as e:
print(f"rejected: {e}")
except NewsdataWebSocketError as e:
print(f"stream error: {e}")All connection options are keyword-only:
ws = NewsDataApiWebSocket(
client,
base_url="wss://ws.newsdata.io/ws/event", # override for staging / self-hosted / proxied
reconnect=True, # auto-reconnect on transient drops; default True
reconnect_delay=1.0, # seconds before first reconnect (doubles each retry)
reconnect_delay_max=30.0, # cap on the reconnect delay
open_timeout=10.0, # handshake timeout (None disables)
ping_interval=20.0, # keepalive ping interval (None disables)
ping_timeout=20.0, # wait for ping reply before dropping (None disables)
additional_headers={"X-Trace": "abc"}, # extra handshake headers
proxy="http://host:port", # proxy URL
)from newsdataapi import (
NewsdataAPIError,
NewsdataAuthError,
NewsdataNetworkError,
NewsdataRateLimitError,
)
try:
client.latest_api(q="news")
except NewsdataAuthError as e:
print(f"bad API key (HTTP {e.status_code})")
except NewsdataRateLimitError as e:
print(f"rate limited; retry after {e.retry_after}s")
except NewsdataAPIError as e:
print(f"API error {e.status_code}: {e.response_body}")
except NewsdataNetworkError as e:
print(f"network failure: {e.original}")The full hierarchy:
NewsdataException
├── NewsdataValidationError (also a ValueError; carries .param)
├── NewsdataAPIError (carries .status_code, .response_body)
│ ├── NewsdataAuthError (401 / 403)
│ ├── NewsdataRateLimitError (429; carries .retry_after)
│ └── NewsdataServerError (5xx)
├── NewsdataNetworkError (carries .original)
└── NewsdataWebSocketError (real-time stream)
└── NewsdataWebSocketAuthError (handshake 401 / 403, or policy-violation close 1008)
NewsdataException is always a valid catch-all.
client.save_to_csv(response, folder_path="./out", filename="latest_news")
# Or set folder_path once on the client and reuse:
client = NewsDataApiClient(apikey, folder_path="./out")
client.save_to_csv(response, filename="latest_news")save_to_csv returns a pathlib.Path. Cell values that are dicts or lists are stringified (key:value,key:value for dicts, comma-joined for lists). Quoting is delegated to the standard csv.DictWriter, so the output round-trips correctly through any CSV reader.
The function is also importable as a standalone:
from newsdataapi import save_to_csv
save_to_csv(response, folder_path="./out", filename="latest_news")client = NewsDataApiClient(
apikey="...",
request_timeout=30, # seconds; default 30
max_retries=5, # default 5
retry_backoff=2.0, # base seconds, exponential; default 2.0
retry_backoff_max=60.0, # cap on a single retry sleep; default 60.0
pagination_delay=1.0, # seconds between pages; default 1.0
max_result=None, # cap on merged results in scroll mode; default None (no cap)
max_pages=None, # cap on pages yielded in paginate mode; default None (no cap)
proxies={"https": "..."}, # passed with every request
accept_language="en", # Accept-Language header
include_headers=False, # if True, returned dicts include response_headers
base_url="...", # override for staging / proxied environments
session=my_session, # inject your own requests.Session
folder_path="./out", # default folder for save_to_csv; default None
)Defaults sleep about a minute total across all retries (2 s → 4 s → 8 s → 16 s → 32 s, capped at 60 s); 429 responses honor Retry-After (both integer-seconds and HTTP-date forms are parsed). The API key is redacted in log output.
This project uses uv for environment and lock management.
git clone https://github.com/newsdataapi/python-client
cd python-client
uv sync # creates .venv, installs runtime + dev deps from uv.lockRun the suite:
uv run pytest # unit tests only (default)
PYTEST_TOKEN=<api-key> uv run pytest -m integration # live-API tests
PYTEST_TOKEN=<api-key> uv run pytest -m "" # all tests
uv run ruff check src/ tests/ examples/
uv run mypy src/Dev dependencies live in PEP 735 [dependency-groups].dev (uv-native). Plain pip install -e ".[dev]" will not pick them up; if you can't use uv, install the contents of the dev group in pyproject.toml by hand.
Official Newsdata.io clients across languages and runtimes:
- Node.js — newsdataapi/newsdata-nodejs-client (npm)
- React (hooks) — newsdataapi/newsdata-reactjs-client (npm)
- PHP — newsdataapi/php-client (Packagist)
- Java — newsdataapi/newsdata-java-sdk (Maven Central)
- .NET — newsdataapi/newsdata-dotnet-sdk (NuGet)
- Go — newsdataapi/newsdata-go-client (pkg.go.dev)
- Dart / Flutter — newsdataapi/newsdata-flutter-client (pub.dev)
- MCP Server (AI assistants) — newsdataapi/newsdata.io-mcp (PyPI)
Also see free news datasets for ML / NLP work.
MIT. See the LICENSE file.