SQL-Alchemy session management

Question

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,
                                         autoflush=True,
                                         bind=engine))

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

def init_db():
    import model
    Base.metadata.create_all(bind=engine)

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

def on_data(self, data):
    try:
        tweet_json = json.loads(data)
        tweet = Tweet(...)
        tweet.save()
    except:
       print_exc()
       db_session.rollback()
    return True

The Tweet.save called above:

def save(self):
    db_session.add(self)
    db_session.commit()

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),
        'last_update':datetime.now()
    })

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:

@classmethod
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})
    db_session.commit()

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