Quickstart recipes

This page contains short, copy-paste-ready recipes for the most common tasks. Each recipe is a complete script - replace YOUR_API_KEY with the key from your dashboard, or set the EVENTREGISTRY_API_KEY environment variable and create the instance with EventRegistry() (see Authorization).

The recipes are written for version 11.0 of the package (pip install --upgrade eventregistry, Python 3.9 or newer).

• Articles about a company in the last 7 days
• The same search with keywords
• Paging, limits and tokens
• Checking token usage before and after a job
• Restricting to sources and languages, skipping duplicates
• Getting the full article body and selected details
• Events about a topic and the articles of one event
• Polling for new articles
• Aggregates: top sources and sentiment
• Handling errors in a long-running job

Articles about a company in the last 7 days

Find the URI of the concept with getConceptUri(), limit the search with dateStart and iterate over the results with QueryArticlesIter. maxItems limits how many articles are downloaded and sortBy = "date" returns the newest first.

import datetime
from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY)

teslaUri = er.getConceptUri("Tesla")     # the best matching concept, or None if nothing matches
print(teslaUri)

weekAgo = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days = 7)
q = QueryArticlesIter(conceptUri = teslaUri, dateStart = weekAgo)

for art in q.execQuery(er, sortBy = "date", maxItems = 200):
    print(art["dateTime"], art["source"]["title"], art["title"])

A search by concept finds the articles about the company in all the languages, also when it is mentioned with a different name. The properties of the returned articles are described in the Article data model.

The same search with keywords

The same search using words instead of a concept. By default a keyword is searched as a phrase in the body of the article.

import datetime
from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY)

weekAgo = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days = 7)
q = QueryArticlesIter(
    # any of the phrases has to be mentioned
    keywords = QueryItems.OR(["Tesla", "Cybertruck", "Model Y"]),
    # look for them in the title and the body
    keywordsLoc = "body,title",
    lang = "eng",
    dateStart = weekAgo)

for art in q.execQuery(er, sortBy = "date", maxItems = 200):
    print(art["dateTime"], art["title"])

Concept or keywords? Use a concept when you are looking for a known thing (a company, a person, a place, a topic that has a Wikipedia page): the search works across languages and scripts, covers different names of the same thing ("Tesla", "Tesla Motors", "Tesla Inc.") and avoids other meanings of the word. Use keywords when you need an exact word or phrase - a product name, a slogan, a rare term, something that is not a concept - or when you want to control where the text appears (keywordsLoc) and how it is matched (keywordSearchMode, see keyword searches). The two can be combined in one query; all the conditions then have to be met. See also Finding concepts for keywords.

Paging, limits and tokens

How many items does a request return?

What	Items per request	Class
articles	at most 100	RequestArticlesInfo(count = 100)
events	at most 50	RequestEventsInfo(count = 50)
mentions	at most 100	RequestMentionsInfo(count = 100)
list of article URIs	at most 50,000 (default 10,000)	RequestArticlesUriWgtList(count = 50000)
list of event URIs	at most 100,000 (default 50,000)	RequestEventsUriWgtList(count = 100000)

How does maxItems map to requests? The iterators (QueryArticlesIter, QueryEventsIter, QueryMentionsIter, QueryEventArticlesIter) download the results page by page while you iterate - always full pages of 100 articles (or mentions) or 50 events. maxItems = 250 for articles therefore makes 3 requests (100 + 100 + 50 used), and maxItems = 250 for events makes 5 requests. Without maxItems the iterator continues until all the matching results are downloaded, which can be a lot of requests - check the number of results first.

count() before iterating. The count() method returns the number of matching results. It is a search itself, so it also uses tokens.

import datetime, math
from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY)
weekAgo = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days = 7)
q = QueryArticlesIter(conceptUri = er.getConceptUri("Tesla"), lang = "eng", dateStart = weekAgo)

total = q.count(er)
print("%d matching articles = %d requests to download them all" % (total, math.ceil(total / 100)))

# download at most the newest 1000 of them (10 requests)
articles = list(q.execQuery(er, sortBy = "date", maxItems = 1000))

If you need a specific page of the results, or want to control the requests yourself, use QueryArticles with RequestArticlesInfo(page = ..., count = 100):

q = QueryArticles(conceptUri = er.getConceptUri("Tesla"), lang = "eng", dateStart = weekAgo)
q.setRequestedResult(RequestArticlesInfo(page = 2, count = 100, sortBy = "date"))
res = er.execQuery(q)
print(res["articles"]["totalResults"], res["articles"]["pages"])
for art in res["articles"]["results"]:
    print(art["title"])

How are tokens consumed? Each request (search) uses tokens of your plan. As described on the plans page, for searches among the recent news articles a search action uses 1 token and retrieves up to 100 articles (or 50 events). Searching the historical data (the archive, content older than about a month) uses more tokens, depending on the number of searched years - the plans page gives the example of an article search over 2015-2017 (3 years of content) that takes 15 tokens. Event searches, aggregates and text analytics have their own prices, listed on the same page. So:

• downloading 1,000 recent articles with an iterator = 10 searches;
• the same over the archive costs more for every one of the 10 searches - limit the dates of the query to what you need;
• if you only need recent content, create the instance with EventRegistry(allowUseOfArchive = False): the queries then never use the archive.

Checking token usage before and after a job

getUsageInfo() returns the number of used and available tokens of your account and printLastReqStats() prints the cost of the last request.

from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY, allowUseOfArchive = False)

before = er.getUsageInfo()
print("used %s of %s tokens" % (before["usedTokens"], before["availableTokens"]))

q = QueryArticlesIter(keywords = "electric vehicles", lang = "eng")
articles = list(q.execQuery(er, sortBy = "date", maxItems = 300))
# tokens used by the last request, the performed action and whether the archive was used
er.printLastReqStats()

after = er.getUsageInfo()
print("the job used %s tokens" % (after["usedTokens"] - before["usedTokens"]))

Restricting to sources and languages, skipping duplicates

Sources can be given by their URI (sourceUri), by the location of the publisher (sourceLocationUri), by a predefined group (sourceGroupUri) or by their ranking. isDuplicateFilter = "skipDuplicates" removes the articles that are copies of other articles.

from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY, allowUseOfArchive = False)

q = QueryArticlesIter(
    conceptUri = er.getConceptUri("artificial intelligence"),
    # only from these sources
    sourceUri = QueryItems.OR([
        er.getNewsSourceUri("bbc"),
        er.getNewsSourceUri("reuters"),
        er.getNewsSourceUri("spiegel")]),
    # in English or German
    lang = QueryItems.OR(["eng", "deu"]),
    # skip the articles that are duplicates of other articles
    isDuplicateFilter = "skipDuplicates")

for art in q.execQuery(er, sortBy = "date", maxItems = 100):
    print(art["lang"], art["source"]["uri"], art["title"])

Other ways of choosing the sources:

# all the sources located in Germany, but only the top ranked ones
q = QueryArticlesIter(
    keywords = "Tesla",
    sourceLocationUri = er.getLocationUri("Germany"),
    startSourceRankPercentile = 0,
    endSourceRankPercentile = 30,
    # everything except these two
    ignoreSourceUri = QueryItems.OR(["bild.de", "focus.de"]))

See Filtering content by news sources and Supported languages.

Getting the full article body and selected details

What is returned for each article is set with ReturnInfo. The full body is returned by default (bodyLen = -1); concepts, categories, links, social shares and other details have to be requested.

from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY, allowUseOfArchive = False)

returnInfo = ReturnInfo(
    articleInfo = ArticleInfoFlags(
        bodyLen = -1,           # the full body (use e.g. 500 for the first 500 characters, 0 for no body)
        concepts = True,        # the concepts mentioned in the article
        categories = True,      # the categories of the article
        links = True,           # the links found in the body
        socialScore = True),    # the number of shares on social media
    # return the labels of the concepts in English and German
    conceptInfo = ConceptInfoFlags(lang = ["eng", "deu"]),
    # add the location and the ranking of the news source
    sourceInfo = SourceInfoFlags(location = True, ranking = True))

q = QueryArticlesIter(conceptUri = er.getConceptUri("Tesla"), lang = "eng")
for art in q.execQuery(er, sortBy = "date", returnInfo = returnInfo, maxItems = 20):
    print(art["title"])
    print(len(art["body"]), "characters")
    print([concept["label"]["eng"] for concept in art["concepts"]])
    print([category["uri"] for category in art["categories"]])

To get the details of articles whose URIs you already have, use QueryArticle:

q = QueryArticle(["247634888", "247634889"])
q.setRequestedResult(RequestArticleInfo(returnInfo = returnInfo))
res = er.execQuery(q)

All the flags are listed on the ReturnInfo page. A misspelled flag raises a TypeError or logs a warning.

Events about a topic and the articles of one event

An event is a group of articles, possibly in several languages, that report about the same thing. Search for events with QueryEventsIter and get the articles of an event with QueryEventArticlesIter.

import datetime
from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY)

monthAgo = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days = 30)
q = QueryEventsIter(
    conceptUri = er.getConceptUri("earthquake"),
    dateStart = monthAgo,
    # only the events reported in at least 20 articles
    minArticlesInEvent = 20)

events = list(q.execQuery(er, sortBy = "size", maxItems = 10))
for event in events:
    print(event["uri"], event["eventDate"], event["totalArticleCount"], event["title"].get("eng"))

# the English articles of the largest event
if events:
    artIter = QueryEventArticlesIter(events[0]["uri"], lang = "eng")
    for art in artIter.execQuery(er, sortBy = "date", maxItems = 50):
        print(art["dateTime"], art["source"]["title"], art["title"])

More details about a single event (its sources, keywords, similar events, ...) are available with QueryEvent. The properties of an event are described in the Event data model.

Polling for new articles

With GetRecentArticles and filters. The feed of new articles is computed once a minute. Call getUpdates() once a minute and you get the articles added since the previous call. Since version 11.0 the conditions can be given as named parameters.

import time
from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY)
recentQ = GetRecentArticles(er,
    conceptUri = QueryItems.OR([er.getConceptUri("Tesla"), er.getConceptUri("Rivian")]),
    lang = "eng",
    isDuplicateFilter = "skipDuplicates",
    maxArticleCount = 200)

starttime = time.time()
while True:
    try:
        for art in recentQ.getUpdates():
            print(art["dateTime"], art["title"])
    except EventRegistryError as ex:
        print("the request failed, will try again in a minute:", ex)
    # wait until the next full minute since the start
    time.sleep(60.0 - ((time.time() - starttime) % 60.0))

With a saved topic page and onlyAfterTm. If you have a topic page saved in your account, ask for the articles added after the time of your previous call. Here you choose the polling interval yourself.

import datetime, time
from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY)
topic = TopicPage(er)
topicUri = "your-topic-page-uri"

lastTm = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(minutes = 15)
while True:
    callTm = datetime.datetime.now(datetime.timezone.utc)
    try:
        res = topic.getArticles(uri = topicUri, onlyAfterTm = lastTm, sortBy = "date", count = 100)
        for art in res.get("articles", {}).get("results", []):
            print(art["dateTime"], art["title"])
        # move the time only when the call succeeded, so that nothing is skipped after a failed call
        lastTm = callTm
    except EventRegistryError as ex:
        print("the request failed, will try again:", ex)
    time.sleep(15 * 60)

The URIs of the topic pages of your account are returned by TopicPages(er).getMyTopicPages(). See also Feed of new articles/events.

Aggregates: top sources and sentiment

Instead of the list of articles, QueryArticles can return a summary of all the matching articles. Set the kind of summary with setRequestedResult(); one request returns one kind.

from eventregistry import *

er = EventRegistry(apiKey = YOUR_API_KEY, allowUseOfArchive = False)
q = QueryArticles(conceptUri = er.getConceptUri("Tesla"), lang = "eng")

# the news sources that published the most matching articles
q.setRequestedResult(RequestArticlesSourceAggr(sourceCount = 20))
res = er.execQuery(q, raiseOnError = True)
for item in res["sourceAggr"]["countsPerSource"]:
    print(item["source"]["title"], item["counts"]["frequency"])

# the distribution of the sentiment of the matching articles
q.setRequestedResult(RequestArticlesSentimentAggr())
res = er.execQuery(q, raiseOnError = True)
print(er.format(res))

# the number of the matching articles per day
q.setRequestedResult(RequestArticlesTimeAggr())
res = er.execQuery(q, raiseOnError = True)
print(er.format(res))

All the available result types (top concepts, categories, keywords, authors, languages, locations, ...) are listed on the Searching for articles and Searching for events pages. Aggregates are computed on a sample of the matching articles when there are many of them. Their token cost differs from the cost of a regular search - see the plans page.

Handling errors in a long-running job

A job that makes many requests should survive short outages, stop when it can't continue (invalid key, no more tokens) and skip the queries that the server rejects. Temporary errors (connection problems, timeouts, status 5xx) are repeated by the package itself - repeatFailedRequestCount sets how many times. Requests rejected because of too many simultaneous requests (429) are repeated until they succeed.

import datetime, logging, time
from eventregistry import *

# show when a failed request is being repeated
logging.basicConfig(level = logging.INFO)

# one instance for the whole job: the requests are made one after another, which keeps
# the job far below the limit of 5 simultaneous requests
er = EventRegistry(apiKey = YOUR_API_KEY, repeatFailedRequestCount = 5)

def saveArticle(art):
    print(art["uri"], art["title"])

weekAgo = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days = 7)
companies = ["Tesla", "Rivian", "Lucid Motors", "BYD"]
for name in companies:
    for attempt in range(3):
        try:
            conceptUri = er.getConceptUri(name)
            if conceptUri is None:
                print("no concept found for", name)
                break
            q = QueryArticlesIter(conceptUri = conceptUri, dateStart = weekAgo)
            for art in q.execQuery(er, sortBy = "date", maxItems = 500):
                saveArticle(art)
            break
        except AccessError as ex:
            # 401 or 403: invalid API key, no more tokens or a disabled account - repeating does not help
            raise SystemExit("stopping the job: %s" % ex.responseText)
        except QueryError as ex:
            # the server rejected this query - continue with the next one
            print("skipping %s: %s" % (name, ex))
            break
        except EventRegistryError as ex:
            if not ex.isTemporary:
                raise
            # the request failed also after it was repeated - wait longer and try the company again
            print("temporary problem (%s), waiting before trying again" % ex)
            time.sleep(ex.retryAfter or 300)

If the job is interrupted in the middle of an iteration and repeated, some articles are downloaded again - use the article uri to detect the ones that you already have. The exceptions, the retries and the logging are described on the Errors and troubleshooting page.