NewsData.io Python Client

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"])

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

# 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"])

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"]

for response in ws.stream(registration_id):
    for article in response["results"]:
        print(article["title"], "-", article["link"])

with NewsDataApiWebSocket(client) as ws:
    for response in ws.stream(registration_id):
        print(response["totalResults"])
        break

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

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}")

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}")

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

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
)