# SQL-Alchemy session management

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:tweetsql@127.0.0.1/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


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

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


The Tweet.save called above:

def save(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.)