Tweepy 4+ Streaming Example

stream.py

import tweepy
from dotenv import load_dotenv
import os

load_dotenv()


class Listener(tweepy.Stream):
    def __init__(self, *args, **kwargs):
        super(Listener, self).__init__(*args)

        # dynamically replace built-in methods with user-defined callbacks
        for arg in kwargs:
            if callable(getattr(self, arg, None)):
                setattr(self, arg, kwargs[arg])


def get_twitter_stream(**kwargs):
    """
    Returns a tweepy.Stream object that can be used to stream Twitter data.

    Keyword arguments:
    - callback functions:
        - on_closed: (response) when the stream is closed
        - on_connect: for when a connection is established
        - on_connect_error: for when a connection error occurs
        - on_disconnect: for when the stream disconnects
        - on_exception: (exception) for when an exception occurs
        - on_keepalive: for when a keepalive is received
        - on_request_error: (status_code) for when a request error occurs
        - on_data: (raw_data) for when a new tweet is received
        - on_status: (status) for when a status update is received
        - on_delete: (status_id, user_id) for when a tweet is deleted
        - on_disconnect_message: (message) for when a disconnect message is received
        - on_limit: (track) for when a limit is reached
        - on_scrub_geo: (user_id) for when a location deletion notice is received
        - on_status_withheld: (notice) for when a status is withheld
        - on_user_withheld: (notice) for when a user is withheld
        - on_warning: (warning) for when a warning occurs

    Returns:
    - tweepy.Stream object
    """
    return Listener(
        os.getenv("TWITTER_CONSUMER_KEY"),
        os.getenv("TWITTER_CONSUMER_SECRET"),
        os.getenv("TWITTER_ACCESS_TOKEN"),
        os.getenv("TWITTER_ACCESS_TOKEN_SECRET"),
        **kwargs,
    )


def on_status(status):
    print("@" + status.user.screen_name + ": " + status.text)


twitter_stream = get_twitter_stream(on_status=on_status)

try:
    print("Start streaming.")
    twitter_stream.sample(languages=["en"])
except KeyboardInterrupt:
    print("Stopped.")
finally:
    print("Done.")
    twitter_stream.disconnect()

The docs at https://docs.tweepy.org/en/v4.4.0/streaming.html explain that "to customize the processing of the stream data, Stream needs to be subclassed," but do not include an example of how to pass a custom callback function without defining it in the listener class.

In this example, you can pass custom callback functions as keyword arguments to the get_twitter_stream helper function - they will be passed to the Listener constructor and will replace any of the in-built callbacks methods.