Skip to content

Commit 59d5431

Browse files
authored
Merge pull request #27 from newsdataapi/feature/websocket
Real-time news over WebSocket
2 parents 7eb3e29 + 5e3c167 commit 59d5431

12 files changed

Lines changed: 1435 additions & 653 deletions

File tree

‎.github/workflows/ci.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ jobs:
2121
strategy:
2222
fail-fast: false
2323
matrix:
24-
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
24+
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
2525
steps:
2626
- uses: actions/checkout@v4
2727

‎README.md‎

Lines changed: 84 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313

1414
</div>
1515

16-
`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.
16+
`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).
1717

1818
## Installation
1919

@@ -27,7 +27,7 @@ If you use [uv](https://github.com/astral-sh/uv):
2727
uv add newsdataapi
2828
```
2929

30-
Supports Python 3.8 through 3.14. The only runtime dependency is `requests`.
30+
Supports Python 3.10 through 3.14. The runtime dependencies are `requests` (REST) and `websockets` (real-time streaming).
3131

3232
## Quickstart
3333

@@ -67,7 +67,7 @@ finally:
6767
| `crypto_count_api(from_date, to_date)` | `/crypto/count` | Aggregate crypto counts |
6868
| `market_count_api(from_date, to_date)` | `/market/count` | Aggregate market counts |
6969

70-
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.
70+
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).
7171

7272
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.
7373

@@ -87,6 +87,83 @@ for page in client.latest_api(q="news", paginate=True, max_pages=5):
8787

8888
`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`.
8989

90+
## Real-time news (WebSocket)
91+
92+
Register a query first — the returned `registration_id` identifies it from then on:
93+
94+
```python
95+
from newsdataapi import NewsDataApiClient, NewsDataApiWebSocket
96+
97+
client = NewsDataApiClient("YOUR_API_KEY")
98+
ws = NewsDataApiWebSocket(client)
99+
response = ws.websocket_register(q="bitcoin", language="en")
100+
registration_id = response["results"]["registration_id"]
101+
```
102+
103+
`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.
104+
105+
Then stream — each yielded response has the familiar `status` / `totalResults` / `results` shape:
106+
107+
```python
108+
for response in ws.stream(registration_id):
109+
for article in response["results"]:
110+
print(article["title"], "-", article["link"])
111+
```
112+
113+
Use it as a context manager to close the connection promptly when you stop early (otherwise it closes when iteration ends):
114+
115+
```python
116+
with NewsDataApiWebSocket(client) as ws:
117+
for response in ws.stream(registration_id):
118+
print(response["totalResults"])
119+
break
120+
```
121+
122+
Inside asyncio applications use `stream_async()` — the same class, same behavior, awaited iteration:
123+
124+
```python
125+
import asyncio
126+
127+
async def main():
128+
async with NewsDataApiWebSocket(client) as ws:
129+
async for response in ws.stream_async(registration_id):
130+
for article in response["results"]:
131+
print(article["title"], "-", article["link"])
132+
133+
asyncio.run(main())
134+
```
135+
136+
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:
137+
138+
```python
139+
from newsdataapi import NewsdataWebSocketAuthError, NewsdataWebSocketError
140+
141+
try:
142+
for response in NewsDataApiWebSocket(client).stream(registration_id):
143+
...
144+
except NewsdataWebSocketAuthError as e:
145+
print(f"rejected: {e}")
146+
except NewsdataWebSocketError as e:
147+
print(f"stream error: {e}")
148+
```
149+
150+
All connection options are keyword-only:
151+
152+
```python
153+
ws = NewsDataApiWebSocket(
154+
client,
155+
base_url="wss://ws.newsdata.io/ws/event", # override for staging / self-hosted / proxied
156+
reconnect=True, # auto-reconnect on transient drops; default True
157+
reconnect_delay=1.0, # seconds before first reconnect (doubles each retry)
158+
reconnect_delay_max=30.0, # cap on the reconnect delay
159+
open_timeout=10.0, # handshake timeout (None disables)
160+
ping_interval=20.0, # keepalive ping interval (None disables)
161+
ping_timeout=20.0, # wait for ping reply before dropping (None disables)
162+
additional_headers={"X-Trace": "abc"}, # extra handshake headers
163+
proxy="http://host:port", # proxy URL
164+
)
165+
```
166+
90167
## Error handling
91168

92169
```python
@@ -118,7 +195,9 @@ NewsdataException
118195
│ ├── NewsdataAuthError (401 / 403)
119196
│ ├── NewsdataRateLimitError (429; carries .retry_after)
120197
│ └── NewsdataServerError (5xx)
121-
└── NewsdataNetworkError (carries .original)
198+
├── NewsdataNetworkError (carries .original)
199+
└── NewsdataWebSocketError (real-time stream)
200+
└── NewsdataWebSocketAuthError (handshake 401 / 403, or policy-violation close 1008)
122201
```
123202

124203
`NewsdataException` is always a valid catch-all.
@@ -154,7 +233,7 @@ client = NewsDataApiClient(
154233
pagination_delay=1.0, # seconds between pages; default 1.0
155234
max_result=None, # cap on merged results in scroll mode; default None (no cap)
156235
max_pages=None, # cap on pages yielded in paginate mode; default None (no cap)
157-
proxies={"https": "..."}, # passed to requests.Session.get
236+
proxies={"https": "..."}, # passed with every request
158237
accept_language="en", # Accept-Language header
159238
include_headers=False, # if True, returned dicts include response_headers
160239
base_url="...", # override for staging / proxied environments

‎examples/websocket_usage.py‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
"""Real-time news streaming example for the newsdataapi SDK.
2+
3+
Run with::
4+
5+
NEWSDATA_API_KEY=<key> python examples/websocket_usage.py
6+
7+
Set NEWSDATA_WS_MODE=async to run the asyncio example instead of the sync one.
8+
9+
Articles are matched by a registered query. If NEWSDATA_REGISTRATION_ID is
10+
set, that query is streamed directly; otherwise the example registers a demo
11+
query (``q="pizza"``) first and prints the resulting ``registration_id`` so
12+
you can reuse it on the next run or remove it later with
13+
``NewsDataApiWebSocket(client).websocket_delete(registration_id)``.
14+
"""
15+
16+
from __future__ import annotations
17+
18+
import asyncio
19+
import os
20+
21+
from newsdataapi import (
22+
NewsDataApiClient,
23+
NewsdataAPIError,
24+
NewsDataApiWebSocket,
25+
NewsdataWebSocketAuthError,
26+
)
27+
28+
29+
def _client_and_registration() -> tuple[NewsDataApiClient, str]:
30+
apikey = os.environ.get("NEWSDATA_API_KEY")
31+
if not apikey:
32+
raise SystemExit(
33+
"Set NEWSDATA_API_KEY in your environment before running this example."
34+
)
35+
client = NewsDataApiClient(apikey)
36+
registration_id = os.environ.get("NEWSDATA_REGISTRATION_ID")
37+
if not registration_id:
38+
registration_id = _register_demo_query(client)
39+
return client, registration_id
40+
41+
42+
def _register_demo_query(client: NewsDataApiClient) -> str:
43+
"""Register a demo query and return its ``registration_id``.
44+
45+
Registering an identical query again answers HTTP 409 with the existing
46+
id in the response body — reuse it instead of failing.
47+
"""
48+
try:
49+
response = NewsDataApiWebSocket(client).websocket_register(q="pizza")
50+
except NewsdataAPIError as exc:
51+
if exc.status_code == 409 and exc.response_body:
52+
registration_id: str = exc.response_body["results"]["registration_id"]
53+
print(f"query already registered; reusing {registration_id}")
54+
return registration_id
55+
raise
56+
registration_id = response["results"]["registration_id"]
57+
print(f'registered demo query q="pizza" -> {registration_id}')
58+
return registration_id
59+
60+
61+
def sync_example() -> None:
62+
client, registration_id = _client_and_registration()
63+
64+
# Iterate to receive matched news as it is published. Transient drops are
65+
# reconnected automatically; press Ctrl+C to stop. The context manager is
66+
# optional — it just closes the socket promptly when the block exits.
67+
with NewsDataApiWebSocket(client) as ws:
68+
for response in ws.stream(registration_id):
69+
for article in response["results"]:
70+
print(article.get("title"), "-", article.get("link"))
71+
72+
73+
async def async_example() -> None:
74+
client, registration_id = _client_and_registration()
75+
76+
# The async counterpart — same behavior, awaited iteration.
77+
async with NewsDataApiWebSocket(client) as ws:
78+
async for response in ws.stream_async(registration_id):
79+
for article in response["results"]:
80+
print(article.get("title"), "-", article.get("link"))
81+
82+
83+
def main() -> None:
84+
try:
85+
if os.environ.get("NEWSDATA_WS_MODE") == "async":
86+
asyncio.run(async_example())
87+
else:
88+
sync_example()
89+
except NewsdataWebSocketAuthError as exc:
90+
# Permanent rejection: bad key, no WebSocket entitlement, unknown
91+
# registration_id, device limit, or exhausted quota.
92+
print(f"connection rejected: {exc}")
93+
except KeyboardInterrupt:
94+
print("\nstopped.")
95+
96+
97+
if __name__ == "__main__":
98+
main()

‎pyproject.toml‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,11 @@ dynamic = ["version"]
55
readme = "README.md"
66
license = { file = "LICENSE" }
77
authors = [{ name = "NewsData.io", email = "contact@newsdata.io" }]
8-
requires-python = ">=3.8"
8+
requires-python = ">=3.10"
99

1010
dependencies = [
1111
"requests>=2.25,<3",
12+
"websockets>=16,<17",
1213
]
1314

1415
keywords = [
@@ -17,13 +18,17 @@ keywords = [
1718
"newsdata-io",
1819
"news-api",
1920
"rest-api",
21+
"websocket",
22+
"websockets",
2023
"api-client",
2124
"python-sdk",
2225
"sdk",
2326
"crypto-news",
2427
"financial-news",
2528
"market-news",
2629
"news-aggregator",
30+
"realtime-news",
31+
"news-stream",
2732
]
2833

2934
classifiers = [
@@ -32,8 +37,6 @@ classifiers = [
3237
"License :: OSI Approved :: MIT License",
3338
"Operating System :: OS Independent",
3439
"Programming Language :: Python :: 3",
35-
"Programming Language :: Python :: 3.8",
36-
"Programming Language :: Python :: 3.9",
3740
"Programming Language :: Python :: 3.10",
3841
"Programming Language :: Python :: 3.11",
3942
"Programming Language :: Python :: 3.12",
@@ -87,7 +90,7 @@ addopts = [
8790

8891
[tool.ruff]
8992
line-length = 100
90-
target-version = "py38"
93+
target-version = "py310"
9194

9295
[tool.ruff.lint]
9396
select = [
@@ -104,6 +107,6 @@ select = [
104107
]
105108

106109
[tool.mypy]
107-
python_version = "3.9"
110+
python_version = "3.10"
108111
strict = true
109112
files = ["src/newsdataapi"]

‎src/newsdataapi/__init__.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,19 +10,25 @@
1010
NewsdataRateLimitError,
1111
NewsdataServerError,
1212
NewsdataValidationError,
13+
NewsdataWebSocketAuthError,
14+
NewsdataWebSocketError,
1315
)
16+
from .websocket import NewsDataApiWebSocket
1417

15-
__version__ = "0.2.3"
18+
__version__ = "0.3.0"
1619

1720
__all__ = [
1821
"NewsDataApiClient",
22+
"NewsDataApiWebSocket",
1923
"NewsdataAPIError",
2024
"NewsdataAuthError",
2125
"NewsdataException",
2226
"NewsdataNetworkError",
2327
"NewsdataRateLimitError",
2428
"NewsdataServerError",
2529
"NewsdataValidationError",
30+
"NewsdataWebSocketAuthError",
31+
"NewsdataWebSocketError",
2632
"__version__",
2733
"save_to_csv",
2834
]

‎src/newsdataapi/client.py‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1120,8 +1120,10 @@ def _request(
11201120
self,
11211121
endpoint: str,
11221122
params: Mapping[str, Any],
1123+
*,
1124+
method: str = "GET",
11231125
) -> dict[str, Any]:
1124-
"""Execute a single GET request, with retries.
1126+
"""Execute a single request (GET by default), with retries.
11251127
11261128
Returns the parsed JSON body on success. Raises a
11271129
:class:`NewsdataException` subclass on permanent failure.
@@ -1135,8 +1137,9 @@ def _request(
11351137

11361138
t0 = time.perf_counter()
11371139
try:
1138-
logger.info("GET %s", _redact_url(full_url))
1139-
response = self._session.get(
1140+
logger.info("%s %s", method, _redact_url(full_url))
1141+
response = self._session.request(
1142+
method,
11401143
full_url,
11411144
proxies=self.proxies,
11421145
timeout=self.request_timeout,

‎src/newsdataapi/constants.py‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,20 @@
1515
COUNT_ENDPOINT = "count"
1616
CRYPTO_COUNT_ENDPOINT = "crypto/count"
1717
MARKET_COUNT_ENDPOINT = "market/count"
18+
WEBSOCKET_REGISTER_ENDPOINT = "websocket/register"
19+
WEBSOCKET_FETCH_ENDPOINT = "websocket/fetch"
20+
WEBSOCKET_DELETE_ENDPOINT = "websocket/delete"
1821

1922
# HTTP defaults.
2023
DEFAULT_REQUEST_TIMEOUT = 30 # seconds
2124
DEFAULT_MAX_RETRIES = 5
2225
DEFAULT_RETRY_BACKOFF = 2.0 # base seconds; doubles each attempt
2326
DEFAULT_RETRY_BACKOFF_MAX = 60.0 # cap on any single retry sleep
2427
PAGINATION_DELAY = 1.0 # seconds slept between pages
28+
29+
# Real-time WebSocket defaults (newsdataapi.NewsDataApiWebSocket).
30+
WS_BASE_URL = "wss://ws.newsdata.io/ws/event"
31+
WS_NEWS_TYPE = "latest" # feed a registered query matches against
32+
WS_POLICY_VIOLATION = 1008 # close code for a permanent connection rejection
33+
WS_RECONNECT_DELAY = 1.0 # seconds before the first reconnect; doubles each retry
34+
WS_RECONNECT_DELAY_MAX = 30.0 # cap on the reconnect delay

‎src/newsdataapi/exceptions.py‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,3 +96,14 @@ def __init__(
9696
) -> None:
9797
super().__init__(message)
9898
self.original = original
99+
100+
101+
class NewsdataWebSocketError(NewsdataException):
102+
"""A real-time WebSocket consumer error
103+
(:class:`newsdataapi.NewsDataApiWebSocket`)."""
104+
105+
106+
class NewsdataWebSocketAuthError(NewsdataWebSocketError):
107+
"""The server rejected the WebSocket connection — bad API key, missing
108+
WebSocket entitlement, unknown ``registration_id``, device limit reached,
109+
or exhausted quota. Not retried."""

0 commit comments

Comments
 (0)