Hello World! I’m back to discuss the backend of the app I’m creating. The backend and frontend of the application both make calls to the database, but they operate independently from each other. The front deals with the user input and putting it into database, but has no direct contact with Twitter. The back end apps relay information between Twitter and the database (stored tweets to send out and replies) but remains untouched by the user. I think of Twitter as the hinge between the two different parts of the app.
This part of the application, send_tweets.py, is responsible for checking the “queues” and sending the tweets in the databases on a schedule according to their associated timestamp. You must first connect CQLEngine to your database using the ip address and keyspace, and sync appropriate models that this:
logging.basicConfig(level=logging.INFO) connection.setup(['127.0.0.1'], "tiktok") logging.info("Connected to tiktok database") sync_table(Tweets) sync_table(Queue) sync_table(Tweets_queue)
I also used logging to track events that happen as the program runs. There are a few logging levels to choose from, and the best choice will differ depending on your use case. I chose ‘INFO’ to ensure that I can watch to make sure things are working as I expected them to. Check here to see more info on logging.
You enter your twitter application keys and tokens accordingly. This part of the app will be grabbing tweet from the database and sending them to the twitter account it has the credentials for, and will therefore need access. I used the Configuration File Parser module to keep my authentication hidden. This is important because I’m pushing all my code up to Github to for others to experiment with and use as a learning tool. The idea would be for them to connect to their own Twitter account. The parser reads all my auth data from a YAML file of sorts. Obviously I don’t include that in the code I push to Github.
parser = SafeConfigParser() parser.read('config.txt') CONSUMER_KEY = parser.get('conf', 'CONSUMER_KEY') CONSUMER_SECRET = parser.get('conf', 'CONSUMER_SECRET') ACCESS_KEY = parser.get('conf', 'ACCESS_KEY') ACCESS_SECRET = parser.get('conf', 'ACCESS_SECRET') auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET) auth.set_access_token(ACCESS_KEY, ACCESS_SECRET) api = tweepy.API(auth) g= Queue.objects.all()
For each one of our Queue objects, we take the ids, and we filter out objects from the Tweet_queue table with a datetime less than the current time. These represent the tweets that are due to be sent out on Twitter.The app will look for tweets in the queues where a time_to_send is less than the current time, and send these tweets out to the twitter feed
for x in g: q= Tweets_queue.objects.filter(Tweets_queue.queue_id == x.id, Tweets_queue.time_to_send < datetime.utcnow())
Now we get the tweet text for each of the tweet_queue objects we’ve filtered out, and update our twitter status with it. Meanwhile, we update the tweet_sent table with our newly tweeted tweet, and delete it from tweets_queue table. This is to help stop things from getting messy. A twitter id is also created by the twitter API once the status has been update. We collect this as status.id and make an entry in the Tweets_sent_by_twitter_id. This will be useful in collecting to replies to this tweet later.
for entry in q: tweet = Tweets.get(id = entry.tweet_id) logging.info("Checking for tweets in for queue_id = %s" % entry.queue_id) status = api.update_status(tweet.tweet) Tweets_sent.create(queue_id = entry.queue_id, time_sent = status.created_at, tweet_id = tweet.id) Tweets_queue.objects(queue_id = entry.queue_id, time_to_send = entry.time_to_send).delete() Tweets_sent_by_twitter_id.create(twitter_id = status.id, queue_id = entry.queue_id, tweet_id = tweet.id) logging.info("Sent tweet")
What about check_for_replies.py?This part of the application will look at the Twitter timeline and capture replies to tweets in specific queues. These will be stored in your database in the queue_tweet_responses table.
First we set up the connection for CQLEngine with the database, including the ip address and keyspace:
logging.basicConfig(level=logging.INFO) connection.setup(['127.0.0.1'], "tiktok") logging.info("Connected to tiktok database")
Next we enter the twitter application keys and tokens such that the webapp may have access to our twitter account.
consumer_key = parser.get('conf', 'CONSUMER_KEY') consumer_secret = parser.get('conf', 'CONSUMER_SECRET') access_token = parser.get('conf', 'ACCESS_KEY') access_token_secret = parser.get('conf', 'ACCESS_SECRET')
The StdOutListener class will be responsible for receiving data receive data from the stream. Twitter offers a set of streaming APIs that enables us to have low latency access to the tweet data on Twitter. A streaming client will be pushing messages from twitter, indicating that tweets and events have occurred. We don’t experience the same overhead as we would with REST polling to pull data.The fill_in_timeline method will determine if the tweet on your timeline is a reply. The metadata for these tweets will be collected into the queue_tweet_responses table.
class StdOutListener(tweepy.StreamListener): ''' Handles data received from the stream. ''' def on_status(self, status): list = Tweets_sent_by_twitter_id.objects.filter(twitter_id = status.in_reply_to_status_id) tweet = list.get() Queue_tweet_responses.create(queue_id = tweet.queue_id , time_received = columns.TimeUUID.from_datetime(status.created_at), tweet_id = tweet.tweet_id, response = status.text, user = status.author.screen_name ) return True
How do we restrict this to just replies and not grab everything on the timeline? I’ve put in a filtration device, the fill_in_timeline function. This uses the Twitter API search function, that can look for tweets on the timeline that contain a specific string of text. I choose ‘CassPopQuiz’ as it is alway in a reply.
def fill_in_timeline(auth): api = tweepy.API(auth) public_tweets = api.search('@CassPopQuiz', count =100) for status in public_tweets:
At this point, I also made sure sure we were only dealing with tweets on the timeline that were replies by ensuring that they had a in_reply_to_status_id property. If not, we can assume that the tweet wasn’t a reply, and is not of use to us here.
if status.in_reply_to_status_id != None: #Ensures the tweet is a reply by limited them to those with a in_reply_to_status_id list = Tweets_sent_by_twitter_id.objects.filter(twitter_id = status.in_reply_to_status_id) tweet = list.get() Queue_tweet_responses.create(queue_id = tweet.queue_id , time_received = status.created_at, tweet_id = tweet.tweet_id, response = status.text, user = status.author.screen_name )
I now use Twitter’s streaming API, using my authentication info and the listener I created as arguments. The listener is responsible for receiving the Twitter will determine what is done with it. In this case if there is a in_reply_to_status_id, and entry is created in the queue_tweet_responses table. I make sure to filter the stream using the user id of my account, as I only want tweet info from my timeline, not the whole wider Twitter universe.
stream = tweepy.Stream(auth, listener) stream.filter(follow=['2815304775'])
I’m pretty happy with what I have so far for my scheduler, as bare bones as it is at this point. I’m looking forward to moving on to Phase 2 with this project and smoothing out the rough edges as well as adding some additional helpful features.For instance I would like to make the scheduler do cron jobs. Once again, any suggestion would be greatly appreciated.