twitterapi.tweets.filtered_stream module

filtered_stream.py: Implementation of Twitter filtered stream endpoint.

“The filtered stream endpoint group enables developers to filter the real-time stream of public Tweets. This endpoint group’s functionality includes multiple endpoints that enable you to create and manage rules, and apply those rules to filter a stream of real-time Tweets that will return matching public Tweets. This endpoint group allows users to listen for specific topics and events in real-time, monitor the conversation around competitions, understand how trends develop in real-time, and much more.” (Source: https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/introduction)

Find the Open API Spec under: https://api.twitter.com/2/openapi.json

Examples

Get all rules:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.tweets.filtered_stream import get_rules

rules = [rule async for rule in get_rules()]

Add a rule:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.tweets.filtered_stream import add_or_delete_rules
from sparta.twitterapi.models.twitter_v2_spec import AddOrDeleteRulesRequest, AddRulesRequest, RuleNoId

rule = '(#test OR @projekt_sparta) -is:retweet'
rule_tag = "Testrule"
addrulesreq = AddOrDeleteRulesRequest(AddRulesRequest(add=[RuleNoId(tag=rule_tag, value=rule)]))
print(await add_or_delete_rules(addrulesreq))

Delete a rule:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.tweets.filtered_stream import add_or_delete_rules
from sparta.twitterapi.models.twitter_v2_spec import AddOrDeleteRulesRequest, DeleteRulesRequest, Delete

ruleid = '123456789'
delrulesreq = AddOrDeleteRulesRequest(DeleteRulesRequest(delete=Delete(ids=[ruleid])))
await add_or_delete_rules(delrulesreq)

Delete all rules:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.tweets.filtered_stream import get_rules, add_or_delete_rules
from sparta.twitterapi.models.twitter_v2_spec import AddOrDeleteRulesRequest, DeleteRulesRequest, Delete

ruleids = [rule.id async for rule in get_rules()]
delrulesreq = AddOrDeleteRulesRequest(DeleteRulesRequest(delete=Delete(ids=ruleids)))
await add_or_delete_rules(delrulesreq)

Connect to stream:

import os
import json
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.tweets.filtered_stream import get_stream

async for tweet_response in get_stream(backfill_minutes=0):
    print(json.dumps(tweet_response.tweet))
    print(json.dumps(tweet_response.includes))

async sparta.twitterapi.tweets.filtered_stream.add_or_delete_rules(rules: AddOrDeleteRulesRequest, dry_run: bool = False) → AddOrDeleteRulesResponse

Add or delete rules from a User’s active rule set. Users can provide unique, optionally tagged rules to add. Users can delete their entire rule set or a subset specified by rule ids or values.

Parameters:

rules (AddOrDeleteRulesRequest) – A Twitter AddOrDeleteRulesObject object.
dry_run (bool, optional) – Dry Run can be used with both the add and delete action, with the expected result given, but without actually taking any action in the system (meaning the end state will always be as it was when the request was submitted). This is particularly useful to validate rule changes. Defaults to False.

Raises:

Exception – Rules cannot be added/deleted due to an http error.

Returns:

Returns an Twitter AddOrDeleteRulesResponse object.

Return type:

AddOrDeleteRulesResponse

async sparta.twitterapi.tweets.filtered_stream.get_rules(ids: List[str] | None = None) → AsyncGenerator[Rule, None]

Returns rules from a User’s active rule set.

Parameters:: ids (List[str]) – A list of Rule IDs.
Raises:: Exception – Cannot get the rules due to an http error.
Returns:: AsyncGenerator that yields Twitter Rule objects.
Return type:: AsyncGenerator[Rule, None]
Yields:: Iterator[AsyncGenerator[Rule, None]] – A Twitter Rule object.

async sparta.twitterapi.tweets.filtered_stream.get_stream(backfill_minutes: int = 5, start_time: datetime | None = None, end_time: datetime | None = None) → AsyncGenerator[TweetResponse, None]

Streams Tweets matching the stream’s active rule set.

Parameters:

backfill_minutes (int, optional) – The number of minutes of backfill requested. Defaults to 5.
start_time (datetime) – The oldest UTC timestamp from which the Tweets will be provided. Timestamp is in second granularity and is inclusive (i.e. 12:00:01 includes the first second of the minute). Defaults to None.
end_time (datetime) – The newest, most recent UTC timestamp to which the Tweets will be provided. Timestamp is in second granularity and is exclusive (i.e. 12:00:01 excludes the first second of the minute). Defaults to None.

Raises:

Exception – Cannot open the stream due to an http error.

Returns:

AsyncGenerator that yields TweetResponses.

Return type:

AsyncGenerator[TweetResponse, None]

Yields:

Iterator[AsyncGenerator[TweetResponse, None]] – A TweetResponse Object.