I have written a couple of scripts for accessing Twitter REST API, and storing data in a Postgres DB for data extraction purposes. I know how to use django ORM, or plain SQL, but this was my first attempt at SQLalchemy. I am not sure if the way I use sessions makes sense.

There two scenarios:

The streaming API script will continuously create new Tweet objects for as long as the script runs.

There is also the refresh_tweet_data, which iterates the existing database tweets and updates some metrics. refresh_user_data operates in a similar way.

For some operations, I had to use update_session while for most db_session seems to work better. All my DB operations are specified in my models class.

Can you spot any obvious mistakes, improvements on the way I am using sessions? The current code is working, I am just not sure if I am doing something silly.

The sessions defined in database.py. db_session and update_session used in the scripts are imported from here:

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('postgresql://tweetsql:[email protected]/tweetsql')
db_session = scoped_session(sessionmaker(autocommit=False,

update_session = scoped_session(sessionmaker(autocommit=True,
Base = declarative_base()
Base.query = db_session.query_property()

def init_db():
    import model

The stream script. on_data is called whenever a new Tweet arrives:

def on_data(self, data):
        tweet_json = json.loads(data)
        tweet = Tweet(...)
    return True

The Tweet.save called above:

def save(self):

refresh_user_data extract. This function is called once when the script is executed. Users are currently 190.000 and increasing:

def refresh_user_followers():
    api = tweepy.API(...)

    for user in User.fetch_users():
        followers_ids = api.followers_ids(user.id)
        User.update_followers(user.id, followers_ids)

The User.update_followers function:

def update_followers(cls, user_id, follower_ids):
    update_session.query(User).filter(User.id == user_id).update({
        'followers': list(follower_ids),
        'follower_count': len(follower_ids),

The function from refresh_tweet_data. This is called once when the script executes:

def refresh_tweet_data():
    tweets_to_update = batch_ids() # get tweets ids not updated within 12 hours
    api = tweepy.API(...)

    while tweets_to_update:
        print('.', end="", flush=True)
        response = api.statuses_lookup(tweets_to_update) # Returns tweets for a list of ids
        for tweet in response:
            Tweet.update_data(tweet.id, tweet)
        tweets_to_update = batch_ids()

The Tweet.update_data:

def update_data(cls, tweet_id, tweet):
    db_session.query(Tweet).filter(Tweet.id == tweet_id).update({
        'favorite_count': tweet.favorite_count,
        'last_update': datetime.now(),
        'json': tweet._json,
        'retweet_count': tweet.retweet_count})

(The refresh scripts will run at the same time as the streaming script.)



Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.