Use the NewsData.io Python client library to integrate NewsData.io API into your Python application without having to make HTTP requests directly. Focus on your data, not the boilerplate.
SOURCE CODE
NewsData.io Python Client RepositoryInstallation
Install the official NewsData.io Python client using pip. We recommend using a virtual environment for your projects.
Terminal
$ pip install newsdataapi
Usage
Import Client
Bring in the core NewsDataApiClient class.
Initialize
Authenticate with your secret API key.
Request
Fetch news with flexible parameters.
main.py
from newsdataapi import NewsDataApiClient
# API key authorization, Initialize the client with your API key
api = NewsDataApiClient (apikey="YOUR_API_KEY")
# You can pass empty or with request parameters (ex. country = "us")
response = api.news_api(q="ronaldo", country="us")You can pass any supported request parameters directly into api.news_api(parameter 1, parameter 2)
Pagination
Handling large datasets is seamless. The Python client allows you to loop through results automatically by tracking the nextPage token.
pagination_example.py
from newsdataapi import NewsDataApiClient
# API key authorization, Initialize the client with your API key
api = NewsDataApiClient (apikey="YOUR_API_KEY")
# You can paginate till last page by providing "page" parameter
page=None
while True:
response = api.news_api(page = page)
page = response.get('nextPage',None)
if not page:
breakReal-Time Streaming API
The NewsData.io Python client also supports the Real-Time Streaming API through WebSockets.
Instead of repeatedly sending API requests to check for new articles, you can register a query once and keep a WebSocket connection open. NewsData.io will then send matching news articles to your application as they become available.
The Python client provides five main operations:
Register - Create a query that defines which news you want to receive.
Fetch - Retrieve the queries already registered on your account.
Stream - Receive matching news articles in real time.
Stream asynchronously - Receive real-time news using Python's asyncio.
Delete - Remove a registered query when you no longer need it.
Before you start
Make sure the NewsData.io Python client is installed:
Terminal
$ pip install newsdataapi
You will also need a NewsData.io API key.
Python
API_KEY = "YOUR_API_KEY"
Your API key will look similar to: pub_xxxxxxxxxxxxxxxxxxxx
For detailed information about the Real-Time Streaming API, see our NewsData.io WebSocket - a beginner-friendly guide.
1. Register a Query
Before you can receive real-time news, you need to register a query.
Registering a query tells NewsData.io which news articles you are interested in. The server stores the query and returns a registration_id. You need this registration_id to start streaming news later.
register_query.py
from newsdataapi import NewsDataApiClient, NewsDataApiWebSocket
API_KEY = "YOUR_API_KEY"
client = NewsDataApiClient(API_KEY)
ws = NewsDataApiWebSocket(client)
response = ws.websocket_register(
q="pizza",
language="en"
)
registration_id = response["results"]["registration_id"]
print("Registered! Your registration_id is:")
print(registration_id)In this example, the query requests English-language news about pizza. You can use other NewsData.io filters when registering a query, such as:
q
country
category
language
domainurl
and other supported parameters
Save the returned registration_id. You will need it when starting the stream.
2. Fetch Registered Queries
If you do not remember your registered query IDs, you can retrieve all registered queries associated with your account.
fetch_queries.py
from newsdataapi import NewsDataApiClient, NewsDataApiWebSocket
API_KEY = "YOUR_API_KEY"
client = NewsDataApiClient(API_KEY)
ws = NewsDataApiWebSocket(client)
response = ws.websocket_fetch()
queries = response["results"]["queries"]
if not queries:
print("You have no registered queries yet.")
else:
print(f"You have {len(queries)} registered query/queries:")
for query in queries:
print(
"-",
query.get("registration_id"),
"- >",
query.get("params")
)The response contains the registered queries and their associated parameters.
3. Stream News in Real Time
Once you have a registration_id, you can start receiving matching articles through a WebSocket connection.
The synchronous streaming method is suitable for standard Python applications.
stream_news.py
from newsdataapi import NewsDataApiClient, NewsDataApiWebSocket
API_KEY = "YOUR_API_KEY"
REGISTRATION_ID = "YOUR_REGISTRATION_ID"
client = NewsDataApiClient(API_KEY)
ws = NewsDataApiWebSocket(client)
print("Connecting... waiting for news (press Ctrl+C to stop)")
with ws:
for response in ws.stream(REGISTRATION_ID):
for article in response["results"]:
print(article.get("title"))
print(article.get("link"))
print()The stream remains open and waits for new articles that match your registered query.
Important
The streaming loop does not end automatically. Press Ctrl+C to stop the program. Real-time news is also not necessarily constant. If your query is very specific, it may take some time before a matching article is published. If your internet connection drops, the SDK automatically reconnects.
Each response follows the same general structure as a NewsData.io API response:
{status: "success"totalResults: 2results: [0: {title: "Example news article"link: "https://example.com/article"}]}The articles are available inside: response["results"]
4. Stream News Using asyncio
If your Python application already uses asyncio and async/await, you can use the asynchronous streaming method. This can be useful for applications such as:
FastAPI applications
Discord bots
asynchronous services
applications handling multiple concurrent tasks
stream_news_async.py
import asyncio
from newsdataapi import NewsDataApiClient, NewsDataApiWebSocket
API_KEY = "YOUR_API_KEY"
REGISTRATION_ID = "YOUR_REGISTRATION_ID"
async def stream_news():
client = NewsDataApiClient(API_KEY)
ws = NewsDataApiWebSocket(client)
print("Connecting (async)... waiting for news (press Ctrl+C to stop)")
async with ws:
async for response in ws.stream_async(REGISTRATION_ID):
for article in response["results"]:
print(article.get("title"))
print(article.get("link"))
print()
asyncio.run(stream_news())The asynchronous version provides the same real-time news and response structure as the synchronous version. The main differences are:
async def
async with
async for
ws.stream_async()
If you are new to Python or are not using asyncio, the regular ws.stream() method is recommended.
5. Delete a Registered Query
When you no longer need a registered query, you can delete it using its registration_id.
delete_query.py
from newsdataapi import NewsDataApiClient, NewsDataApiWebSocket
API_KEY = "YOUR_API_KEY"
REGISTRATION_ID = "YOUR_REGISTRATION_ID"
client = NewsDataApiClient(API_KEY)
ws = NewsDataApiWebSocket(client)
ws.websocket_delete(REGISTRATION_ID)
print("Deleted:", REGISTRATION_ID)After the registration is deleted, you can no longer use that registration_id to stream news.
Handling Errors
The Python client provides specific exceptions for common WebSocket and API errors. Import the exceptions:
exceptions.py
from newsdataapi import (
NewsDataApiClient,
NewsdataAPIError,
NewsDataApiWebSocket,
NewsdataWebSocketAuthError,
)NewsdataWebSocketAuthError
NewsdataWebSocketAuthError indicates that the WebSocket connection was refused. Common causes include:
Invalid API key
A plan without WebSocket access
Invalid or non-existent registration_id
Retrying will not resolve an authentication or authorization problem. Check your API key, plan, and registration ID.
Example:
auth_error.py
try:
# Start your WebSocket connection here
pass
except NewsdataWebSocketAuthError as error:
print("Connection refused:", error)
print("Check your API key, plan, and registration_id.")NewsdataAPIError
NewsdataAPIError is used for errors returned by regular API operations such as registering, fetching, or deleting queries.
For example, if you try to register the exact same query again, the API may return status code 409. The error response can contain the existing registration_id, allowing you to reuse the already registered query.
register_safely.py
from newsdataapi import (
NewsDataApiClient,
NewsDataApiWebSocket,
NewsdataAPIError,
)
def register_safely():
client = NewsDataApiClient("YOUR_API_KEY")
ws = NewsDataApiWebSocket(client)
try:
response = ws.websocket_register(
q="pizza",
language="en"
)
registration_id = response["results"]["registration_id"]
print("Registered:", registration_id)
except NewsdataAPIError as error:
if error.status_code == 409:
registration_id = (
error.response_body["results"]["registration_id"]
)
print(
"Already registered, reusing:",
registration_id
)
else:
raise
return registration_idComplete Example
The following example puts the main WebSocket operations together in one Python file.
websocket_example.py
import asyncio
from newsdataapi import (
NewsDataApiClient,
NewsdataAPIError,
NewsDataApiWebSocket,
NewsdataWebSocketAuthError,
)
API_KEY = "YOUR_API_KEY"
REGISTRATION_ID = "PASTE_YOUR_REGISTRATION_ID_HERE"
def make_websocket():
client = NewsDataApiClient(API_KEY)
return NewsDataApiWebSocket(client)
def register_query():
ws = make_websocket()
response = ws.websocket_register(
q="pizza",
language="en"
)
registration_id = response["results"]["registration_id"]
print("Registered:", registration_id)
return registration_id
def fetch_queries():
ws = make_websocket()
response = ws.websocket_fetch()
queries = response["results"]["queries"]
if not queries:
print("No registered queries found.")
return
for query in queries:
print(
query.get("registration_id"),
"- >",
query.get("params")
)
def stream_news(registration_id):
ws = make_websocket()
print("Waiting for news...")
with ws:
for response in ws.stream(registration_id):
for article in response["results"]:
print(article.get("title"))
print(article.get("link"))
print()
async def stream_news_async(registration_id):
ws = make_websocket()
print("Waiting for news asynchronously...")
async with ws:
async for response in ws.stream_async(registration_id):
for article in response["results"]:
print(article.get("title"))
print(article.get("link"))
print()
def delete_query(registration_id):
ws = make_websocket()
ws.websocket_delete(registration_id)
print("Deleted:", registration_id)
if __name__ == "__main__":
try:
# Register a query
# registration_id = register_query()
# Fetch registered queries
# fetch_queries()
# Stream news
# stream_news(REGISTRATION_ID)
# Stream news asynchronously
# asyncio.run(stream_news_async(REGISTRATION_ID))
# Delete a registered query
# delete_query(REGISTRATION_ID)
pass
except NewsdataWebSocketAuthError as error:
print("Connection refused:", error)
print("Check your API key, plan, and registration_id.")
except KeyboardInterrupt:
print("\nStopped.")Uncomment one operation at a time and run the file:
Terminal
$ python your_file.py
For your first test, register a query and then use the returned registration_id to start streaming.
What's Next?
The Python client makes it easy to use both the standard NewsData.io API and real-time news streaming directly from Python. For a detailed explanation of how the Real-Time Streaming API works, including WebSocket registration, streaming, filtering, and practical examples, read our NewsData.io WebSocket - a beginner-friendly guide
You can also explore the NewsData.io Python Client on GitHub
Was this page helpful?