SQLAlchemy で データを PostgreSQL に保存するアプリを組んでみた

Python PostgreSQL SQLAlchemy Twitter

個人的に SQLAlchemy を利用してみたかったのと、最近 PostgreSQL が流行っていそうなので触ろうかと思って、 Twitter の TL からデータを取得して、そのデータを保存するアプリを組んでみました。

最終的にこの程度のとても小さなアプリだけど、 SQLAlchemy が簡便で使い易くてよかった。

また alembic という SQLAlchemy 向けの migration 、changeset 毎に SQLAlchemy で組んだモデルの違いを感知し DB のスキーマを更新するアプリがあり、そろっている感じがしていいなと思った。

事前準備

事前に以下のソフトウェアをインストールした。( Debian の場合 )

モデル定義

declarative_base() が肝

the declarative_base() callable returns a new base class from which all mapped classes should inherit

# -*- coding: utf-8 -*-
from sqlalchemy import BigInteger, Integer, String, Column
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()


class Tweet(Base):
    __tablename__ = 'tweet'

    tweet_id = Column(BigInteger, primary_key=True)
    user_id = Column(Integer)
    text = Column(String)

    def __init__(self, tweet_id, user_id, text):
        self.tweet_id = tweet_id
        self.user_id = user_id
        self.text = text

    def __repr__(self):
        return (
            "<Tweet({0} '{1}')>"
            .format(self.usr_id, self.text))

データアクセス部分

session を通じて DB を操作。

# -*- coding: utf-8 -*-
from sqlalchemy import create_engine
from sqlalchemy.orm import (
    scoped_session,
    sessionmaker,
)

from datadig import conf, models


DBSession = scoped_session(sessionmaker())


def get_session():
    # DB 初期化前なら初期化する
    if not conf.get_conf('db_initialized'):
        db_url = conf.get_conf('db_url')
        engine = create_engine(db_url)
        DBSession.configure(bind=engine)

        # 初期化済を示すためconfig に適当に文字列を保存
        conf.set_conf('db_initialized', 'ok')
    return DBSession()


def store_tweets(tweets=[]):
    """入力された値を DB に保存する"""
    if tweets:
        session = get_session()
        for tweet in tweets:
            _tweet = models.Tweet(*tweet)
            session.add(_tweet)
        session.commit()

まとめ

上記で書いたコードは以下で公開しているのでよかったら。なお、ライセンスは CC0。

https://github.com/jptomo/datadig/

参考