SQLAlchemy: The ORM

July 23, 2011 | View Comments

Long story short; SQLAlchemy is an ORM for Python. SQLAlchemy is far more complex than Django's ORM. You can actually call stored functions in SQLAlchemy. For a Django vs. SQLAlchemy article, have a look at Armin Ronacher's post.

There are two different ways to define a new table. One is; defining a table within the MetaData. Second one is; creating table, class and mapper with Declarative Base. I'm more like a declarative_base guy and I must say that I've never used MetaData.

Warming Up The Engine

This might be a way to define the SQLAlchemy engine. In SQLAlchemy, everything is done via a Session().

from sqlalchemy import (Column, SmallInteger, Unicode, UnicodeText,
    CHAR, VARCHAR, DateTime, ForeignKey, create_engine)
from sqlalchemy.orm import (scoped_session, sessionmaker, relation, relationship)
from sqlalchemy.ext.declarative_base import declarative_base

First, we imported data types such as SmallInteger, Unicode, VARCHAR etc. In the second line, we imported sessionmaker and relations which we use in our tables since SQLAlchemy won't wave its magic wand to define relations between tables. Third line; the declarative_base.

engine = create_engine(ENGINE, echo=True, pool_recycle=3600)
Base = declarative_base()
Session = scoped_session(sessionmaker(bind=engine, autoflush=True,
    autocommit=True, expire_on_commit=True))
session = Session()

Above code defines the engine, and declarative_base class and the session. Once the setup is complete, we need to create the table classes.

class User(Base):
    __tablename__ = 'user'
    __table_args__ = {'mysql_charset': 'utf8', 'mysql_engine': 'InnoDB'}

    uid = Column(CHAR(32), primary_key=True)
    name = Column(Unicode(255), nullable=False)
    username = Column(VARCHAR(32), nullable=False, unique=True)
    password = Column(CHAR(64), nullable=False, index=True)
    registered = Column(DateTime, default=datetime.utcnow())

    def __init__(self, name, username, password):
        self.uid = uuid4().hex
        self.name = name
        self.username = username
        self.password = password

    def __repr__(self):
        return u'<User (%s, %s)>' % (self.uid, self.username)

Here, we define a new class named User. Remember, SQLAlchemy expects you to explicitly define tablenames and all table options so we define the tablename as user.

Then we defined the table options. In table_args we explicitly stated to SQLAlchemy that this table uses utf8 as charset and it uses InnoDB as the engine.

Then we defined the table fields. nullable=False says this field won't accept NULL. unique=True says this field is unique indexed. default=datetime.utcnow() says field's default value.

Many-To-Many, Many-To-One

Let's say we've simple blog where users' post their articles.

class Article(Base):
    __tablename__ = 'article'
    __table_args__ = {'mysql_charset': 'utf8', 'mysql_engine': 'InnoDB'}

    aid = Column(CHAR(32), primary_key=True)
    author_id = Column(CHAR(32), ForeignKey('user.uid', ondelete='cascade'))
    title = Column(Unicode(255), nullable=False)
    content = Column(UnicodeText, nullable=False)
    posted = Column(DateTime, default=datetime.utcnow())

    author = relation(User, primaryjoin=User.uid == author_id, lazy='select')

    # Let's assume there's also a Comments table.
    # In this case we could make a relation between tables by ordering the results.
    comments = relation('Comments', order_by='Comments.posted.desc()', backref='article')

In the above example, we create a relation to User table. To create a many-to-many tables, we need to create the classes and the necessary association class.

CRUD Operations

CRUD is simple.

Create

For example, let's insert some data to Article class.

def insert_article(args):
    res = Article(**args)
    session.add(res)
    session.commit()
    return res

Retrieve

For example; let's retrieve an article with the specified article_id.

def retrieve_article(article_id):
    res = None
    try:
        res = session.query(Article).filter(Article.aid == article_id).first()
    except (NoResultFound):
        session.rollback()
    return res

Another example which we return some articles of a user.

def retrieve_user_articles(user_id):
    res = None
    try:
        res = session.query(Article).filter(Article.author_id == user_id).order_by(Article.posted.desc()).all()
    except (NoResultFound):
        session.rollback()
    return res

Update

def update_article(article_id):
    res = session.query(Article).get(article_id)
    res.title = 'New Title'
    session.add(res)
    session.commit()
    return res

Delete

For example, let's delete an article with specified article_id.

def delete_article(article_id):
    try:
        res = session.query(Article).filter(Article.aid == article_id).delete(synchronize_session='fetch')
        session.commit()
    except (NoResultFound):
        session.rollback()
    return res

SQL/Stored Functions

Calling SQL functions such as MySQL's COUNT() or your own stored functions is very easy with SQLAlchemy.

Let's say we want to count the number of rows in the table.

from sqlalchemy.sql import func

def count_rows():
  res = session.query(Article.aid, func.count(Article.aid).label('rows')).all()
  return res

label stands for SQL's AS (for example; SELECT value AS data FROM table. label is the AS data part). Calling your stored functions is precisely the same as calling SQL functions. Assume that we have a stored function named my_func. In this case, the proper SQLAlchemy code would be; func.my_func(arg).

I didn't cover joins and some other more SQLAlchemy features but SQLAlchemy is an endless sea.