Real-time news over WebSocket

.github/workflows/ci.yml

@@ -21,7 +21,7 @@ jobs:
 strategy:
 fail-fast: false
 matrix:
-python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
 steps:
  - uses: actions/checkout@v4
 

README.md

@@ -13,7 +13,7 @@
 
 </div>
 
-`newsdataapi` is the official Python SDK for the [NewsData.io](https://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.
+`newsdataapi` is the official Python SDK for the [NewsData.io](https://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).
 
 ## Installation
 
@@ -27,7 +27,7 @@ If you use [uv](https://github.com/astral-sh/uv):
 uv add newsdataapi
 ```
 
-Supports Python 3.8 through 3.14. The only runtime dependency is `requests`.
+Supports Python 3.10 through 3.14. The runtime dependencies are `requests` (REST) and `websockets` (real-time streaming).
 
 ## Quickstart
 
@@ -67,7 +67,7 @@ finally:
 | `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.
+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](https://newsdata.io/documentation) — or the [OpenAPI 3.1 spec](https://newsdata.io/openapi.json) — for the full parameter reference.
 
@@ -87,6 +87,83 @@ for page in client.latest_api(q="news", paginate=True, max_pages=5):
 
 `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`.
 
+## Real-time news (WebSocket)
+
+Register a query first — the returned `registration_id` identifies it from then on:
+
+```python
+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:
+
+```python
+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):
+
+```python
+with NewsDataApiWebSocket(client) as ws:
+ for response in ws.stream(registration_id):
+ print(response["totalResults"])
+ break
+```
+
+Inside asyncio applications use `stream_async()` — the same class, same behavior, awaited iteration:
+
+```python
+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:
+
+```python
+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:
+
+```python
+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
+)
+```
+
 ## Error handling
 
 ```python
@@ -118,7 +195,9 @@ NewsdataException
 │ ├── NewsdataAuthError (401 / 403)
 │ ├── NewsdataRateLimitError (429; carries .retry_after)
 │ └── NewsdataServerError (5xx)
-└── NewsdataNetworkError (carries .original)
+├── NewsdataNetworkError (carries .original)
+└── NewsdataWebSocketError (real-time stream)
+ └── NewsdataWebSocketAuthError (handshake 401 / 403, or policy-violation close 1008)
 ```
 
 `NewsdataException` is always a valid catch-all.
@@ -154,7 +233,7 @@ client = NewsDataApiClient(
  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 to requests.Session.get
+ 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

examples/websocket_usage.py

@@ -0,0 +1,98 @@
+"""Real-time news streaming example for the newsdataapi SDK.
+
+Run with::
+
+ NEWSDATA_API_KEY=<key> python examples/websocket_usage.py
+
+Set NEWSDATA_WS_MODE=async to run the asyncio example instead of the sync one.
+
+Articles are matched by a registered query. If NEWSDATA_REGISTRATION_ID is
+set, that query is streamed directly; otherwise the example registers a demo
+query (``q="pizza"``) first and prints the resulting ``registration_id`` so
+you can reuse it on the next run or remove it later with
+``NewsDataApiWebSocket(client).websocket_delete(registration_id)``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+
+from newsdataapi import (
+ NewsDataApiClient,
+ NewsdataAPIError,
+ NewsDataApiWebSocket,
+ NewsdataWebSocketAuthError,
+)
+
+
+def _client_and_registration() -> tuple[NewsDataApiClient, str]:
+ apikey = os.environ.get("NEWSDATA_API_KEY")
+ if not apikey:
+ raise SystemExit(
+ "Set NEWSDATA_API_KEY in your environment before running this example."
+ )
+ client = NewsDataApiClient(apikey)
+ registration_id = os.environ.get("NEWSDATA_REGISTRATION_ID")
+ if not registration_id:
+ registration_id = _register_demo_query(client)
+ return client, registration_id
+
+
+def _register_demo_query(client: NewsDataApiClient) -> str:
+ """Register a demo query and return its ``registration_id``.
+
+ Registering an identical query again answers HTTP 409 with the existing
+ id in the response body — reuse it instead of failing.
+ """
+ try:
+ response = NewsDataApiWebSocket(client).websocket_register(q="pizza")
+ except NewsdataAPIError as exc:
+ if exc.status_code == 409 and exc.response_body:
+ registration_id: str = exc.response_body["results"]["registration_id"]
+ print(f"query already registered; reusing {registration_id}")
+ return registration_id
+ raise
+ registration_id = response["results"]["registration_id"]
+ print(f'registered demo query q="pizza" -> {registration_id}')
+ return registration_id
+
+
+def sync_example() -> None:
+ client, registration_id = _client_and_registration()
+
+ # Iterate to receive matched news as it is published. Transient drops are
+ # reconnected automatically; press Ctrl+C to stop. The context manager is
+ # optional — it just closes the socket promptly when the block exits.
+ with NewsDataApiWebSocket(client) as ws:
+ for response in ws.stream(registration_id):
+ for article in response["results"]:
+ print(article.get("title"), "-", article.get("link"))
+
+
+async def async_example() -> None:
+ client, registration_id = _client_and_registration()
+
+ # The async counterpart — same behavior, awaited iteration.
+ async with NewsDataApiWebSocket(client) as ws:
+ async for response in ws.stream_async(registration_id):
+ for article in response["results"]:
+ print(article.get("title"), "-", article.get("link"))
+
+
+def main() -> None:
+ try:
+ if os.environ.get("NEWSDATA_WS_MODE") == "async":
+ asyncio.run(async_example())
+ else:
+ sync_example()
+ except NewsdataWebSocketAuthError as exc:
+ # Permanent rejection: bad key, no WebSocket entitlement, unknown
+ # registration_id, device limit, or exhausted quota.
+ print(f"connection rejected: {exc}")
+ except KeyboardInterrupt:
+ print("\nstopped.")
+
+
+if __name__ == "__main__":
+ main()

pyproject.toml

@@ -5,10 +5,11 @@ dynamic = ["version"]
 readme = "README.md"
 license = { file = "LICENSE" }
 authors = [{ name = "NewsData.io", email = "contact@newsdata.io" }]
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 
 dependencies = [
  "requests>=2.25,<3",
+ "websockets>=16,<17",
 ]
 
 keywords = [
@@ -17,13 +18,17 @@ keywords = [
  "newsdata-io",
  "news-api",
  "rest-api",
+ "websocket",
+ "websockets",
  "api-client",
  "python-sdk",
  "sdk",
  "crypto-news",
  "financial-news",
  "market-news",
  "news-aggregator",
+ "realtime-news",
+ "news-stream",
 ]
 
 classifiers = [
@@ -32,8 +37,6 @@ classifiers = [
  "License :: OSI Approved :: MIT License",
  "Operating System :: OS Independent",
  "Programming Language :: Python :: 3",
- "Programming Language :: Python :: 3.8",
- "Programming Language :: Python :: 3.9",
  "Programming Language :: Python :: 3.10",
  "Programming Language :: Python :: 3.11",
  "Programming Language :: Python :: 3.12",
@@ -87,7 +90,7 @@ addopts = [
 
 [tool.ruff]
 line-length = 100
-target-version = "py38"
+target-version = "py310"
 
 [tool.ruff.lint]
 select = [
@@ -104,6 +107,6 @@ select = [
 ]
 
 [tool.mypy]
-python_version = "3.9"
+python_version = "3.10"
 strict = true
 files = ["src/newsdataapi"]

src/newsdataapi/__init__.py

@@ -10,19 +10,25 @@
 NewsdataRateLimitError,
 NewsdataServerError,
 NewsdataValidationError,
+NewsdataWebSocketAuthError,
+NewsdataWebSocketError,
 )
+from .websocketimportNewsDataApiWebSocket
 
-__version__="0.2.3"
+__version__="0.3.0"
 
 __all__= [
 "NewsDataApiClient",
+"NewsDataApiWebSocket",
 "NewsdataAPIError",
 "NewsdataAuthError",
 "NewsdataException",
 "NewsdataNetworkError",
 "NewsdataRateLimitError",
 "NewsdataServerError",
 "NewsdataValidationError",
+"NewsdataWebSocketAuthError",
+"NewsdataWebSocketError",
 "__version__",
 "save_to_csv",
 ]

src/newsdataapi/client.py

@@ -1120,8 +1120,10 @@ def _request(
  self,
  endpoint: str,
  params: Mapping[str, Any],
+ *,
+ method: str = "GET",
  ) -> dict[str, Any]:
- """Execute a single GET request, with retries.
+ """Execute a single request (GET by default), with retries.
 
  Returns the parsed JSON body on success. Raises a
  :class:`NewsdataException` subclass on permanent failure.
@@ -1135,8 +1137,9 @@ def _request(
 
  t0 = time.perf_counter()
  try:
- logger.info("GET %s", _redact_url(full_url))
- response = self._session.get(
+ logger.info("%s %s", method, _redact_url(full_url))
+ response = self._session.request(
+ method,
  full_url,
  proxies=self.proxies,
  timeout=self.request_timeout,

src/newsdataapi/constants.py

@@ -15,10 +15,20 @@
 COUNT_ENDPOINT = "count"
 CRYPTO_COUNT_ENDPOINT = "crypto/count"
 MARKET_COUNT_ENDPOINT = "market/count"
+WEBSOCKET_REGISTER_ENDPOINT = "websocket/register"
+WEBSOCKET_FETCH_ENDPOINT = "websocket/fetch"
+WEBSOCKET_DELETE_ENDPOINT = "websocket/delete"
 
 # HTTP defaults.
 DEFAULT_REQUEST_TIMEOUT = 30 # seconds
 DEFAULT_MAX_RETRIES = 5
 DEFAULT_RETRY_BACKOFF = 2.0 # base seconds; doubles each attempt
 DEFAULT_RETRY_BACKOFF_MAX = 60.0 # cap on any single retry sleep
 PAGINATION_DELAY = 1.0 # seconds slept between pages
+
+# Real-time WebSocket defaults (newsdataapi.NewsDataApiWebSocket).
+WS_BASE_URL = "wss://ws.newsdata.io/ws/event"
+WS_NEWS_TYPE = "latest" # feed a registered query matches against
+WS_POLICY_VIOLATION = 1008 # close code for a permanent connection rejection
+WS_RECONNECT_DELAY = 1.0 # seconds before the first reconnect; doubles each retry
+WS_RECONNECT_DELAY_MAX = 30.0 # cap on the reconnect delay

src/newsdataapi/exceptions.py

@@ -96,3 +96,14 @@ def __init__(
  ) -> None:
  super().__init__(message)
  self.original = original
+
+
+class NewsdataWebSocketError(NewsdataException):
+ """A real-time WebSocket consumer error
+ (:class:`newsdataapi.NewsDataApiWebSocket`)."""
+
+
+class NewsdataWebSocketAuthError(NewsdataWebSocketError):
+ """The server rejected the WebSocket connection — bad API key, missing
+ WebSocket entitlement, unknown ``registration_id``, device limit reached,
+ or exhausted quota. Not retried."""