twitterapi.compliance.compliance_stream module

compliance_stream.py: Implementation of Twitter compliance stream endpoint.

“Twitter is committed to our community of developers who build with the Twitter API. As part of this commitment, we aim to make our API open and fair to developers, safe for people on Twitter, and beneficial for the Twitter platform as a whole. It is crucial that any developer who stores Twitter content offline, ensures the data reflects user intent and the current state of content on Twitter. For example, when someone on Twitter deletes a Tweet or their account, protects their Tweets, edits a Tweet, or scrubs the geo information from their Tweets, it is critical for both Twitter and our developers to honor that person’s expectations and intent.” (Source: https://developer.twitter.com/en/docs/twitter-api/compliance/streams/introduction)

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

Examples

Start the compliance stream for tweets:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.compliance.compliance_stream import get_tweet_compliance_stream

async for complianceEvent in get_tweet_compliance_stream(partition=1):
    print(complianceEvent)

Start the compliance stream for users:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.compliance.compliance_stream import get_user_compliance_stream

async for complianceEvent in get_user_compliance_stream(partition=1):
    print(complianceEvent)
async sparta.twitterapi.compliance.compliance_stream.get_tweet_compliance_stream(partition: int = 1, backfill_minutes: int | None = None, start_time: datetime | None = None, end_time: datetime | None = None) → AsyncGenerator[TweetComplianceStreamResponse1, None]

Get an asynchronous tweet compliance-stream as generator.

Parameters:

  • partition (int, optional) – Must be set to 1, 2, 3 or 4. Tweet compliance events are split across 4 partitions, so 4 separate streams are needed to receive all events. Defaults to 1.

  • backfill_minutes (Optional[int], optional) – The number of minutes of backfill requested. By passing this parameter, you can recover up to five minutes worth of data that you might have missed during a disconnection. Defaults to None.

  • start_time (Optional[datetime], optional) – The earliest UTC timestamp from which the Tweet Compliance events will be provided. Defaults to None.

  • end_time (Optional[datetime], optional) – The latest UTC timestamp to which the Tweet Compliance events will be provided. Defaults to None.

Raises:

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

Returns:

AsyncGenerator that yields Twitter TweetComplianceStreamResponse1 objects.

Return type:

AsyncGenerator[TweetComplianceStreamResponse1, None]

Yields:

Iterator[AsyncGenerator[TweetComplianceStreamResponse1, None]] – A Twitter TweetComplianceStreamResponse1 object.

async sparta.twitterapi.compliance.compliance_stream.get_user_compliance_stream(partition: int = 1, backfill_minutes: int | None = None, start_time: datetime | None = None, end_time: datetime | None = None) → AsyncGenerator[UserComplianceStreamResponse1, None]

Get an asynchronous user compliance-stream as generator.

Parameters:

  • partition (int, optional) – Must be set to 1, 2, 3 or 4. User compliance events are split across 4 partitions, so 4 separate streams are needed to receive all events. Defaults to 1.

  • backfill_minutes (Optional[int], optional) – The number of minutes of backfill requested. By passing this parameter, you can recover up to five minutes worth of data that you might have missed during a disconnection. Defaults to None.

  • start_time (Optional[datetime], optional) – The earliest UTC timestamp from which the Tweet Compliance events will be provided. Defaults to None.

  • end_time (Optional[datetime], optional) – The latest UTC timestamp to which the Tweet Compliance events will be provided. Defaults to None.

Raises:

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

Returns:

AsyncGenerator that yields Twitter UserComplianceStreamResponse1 objects.

Return type:

AsyncGenerator[UserComplianceStreamResponse1, None]

Yields:

Iterator[AsyncGenerator[UserComplianceStreamResponse1, None]] – A Twitter UserComplianceStreamResponse1 object.