Tweets hashtags real-time processing with Faust - Command

command.py

@app.command(option("--track", type=str, help="List of words in CSV format to filter stream of tweets", required=True))
async def hashtags_events_generator(self, track: str):
    """
    Generate Kafka events for each hashtag found in tweets matching with the track used to filter the stream.
    """
    logger.info(f"Searching tweets with hashtags: {track}")

    client = _get_twitter_client()
    req = client.stream.statuses.filter.post(track=track)

    async with req as stream:
        async for tweet in stream:
            tweet_hashtags = _get_hashtags_from_tweet(tweet=tweet)
            await _produce_events_for_hashtags(hashtags=tweet_hashtags)