Getting Started With Sqlalchemy Orm

Aug 14th, 2020 - written by Kimserey with .

In a previous post, we talked about Alembic which is a migration tool built on top of SQLAlchemy. SQLAlchemy is a Python SQL toolkit and Object Relational Mapper (ORM). In today’s post, we will look at how the ORM part can be leverage to easily create and manage a simple database schema.

Create an Engine
Database Models
Insert
Query

This post is heavily inspired from the official SQLAlchemy ORM Tutorial.

Create an Engine

SQLAlchemy supports all major SQL engines, Postgres, MySQL, Oracle, MSSQL, SQLite, etc… In this post we will be using SQLite as it provides an in memory storage, quick and easy for prototyping. We start by creating the engine and specifying our connection string:

from sqlalchemy import create_engine

engine = create_engine("sqlite:///:memory:", echo=True)

echo=True will echo all SQL commands.

Database Models

We then define our database models:

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    profile = relationship("UserProfile", uselist=False, back_populates="user")

    def __repr__(self):
        return "<User(id={}, name={}>".format(self.id, self.name)


class UserProfile(Base):
    __tablename__ = "user_profile"

    user_id = Column(Integer, ForeignKey("user.id"), primary_key=True)
    user = relationship("User", back_populates="profile")

    job = Column(String)
    address = Column(String)

    def __repr__(self):
        return "<UserProfile(user_id={}, job={}, address={})>".format(
            self.user_id, self.job, self.address
        )

declarative_base creates a base class, which here we call Base, which contains the catalog of clases and tables in a metadata property. Each table mapping defined can be retrived from Base.metadata.tables.

> Base.metadata.tables

immutabledict({'user': Table('user', MetaData(bind=None), Column('id', Integer(), table=<user>, primary_key=True, nullable=False), Column('name', String(), table=<user>), schema=None), 'user_profile': Table('user_profile', MetaData(bind=None), Column('user_id', Integer(), ForeignKey('user.id'), table=<user_profile>, primary_key=True, nullable=False), Column('job', String(), table=<user_profile>), Column('address', String(), table=<user_profile>), schema=None)})

Here we defined User and UserProfile tables by giving them a name using the __tablename__ attribute and defining columns with Column.

In order to define an object property relationship, we use relationship from sqlalchemy.orm. In User, profile will not be a column in the table but rather a relationship with UserProfile. Similarly, UserProfile.user will be a relationship with User. SQLAlchemy knows how to use the existing foreign key on user to make either a lazy select, or an eager join.

Once we’re done creating our model, we can create the database schema by executing:

Base.metadata.create_all(engine)

By looking at the logs, we should be able to spot the CREATE TABLE queries indicating that our tables were created.

Insert

Now that we have our database models and database schema created, we can start inserting data. To start interacting with the database, we need to get hold of a session.

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

A Session class is created using the sessionmaker which generates a new class with the settings provided to the sessionmaker like autoflush or autocommit. Here we directly bind the Session with our SQLite engine. Doing that we can be sure that when we instantiate sessions, we get a session connecting to our SQLite engine.

We then create a few users:

from faker import Faker

# Faker is used to auto generate comprehensible example of name/job/address.
fake = Faker()


def create_user():
    user = User(name=fake.name())
    user.profile = UserProfile(job=fake.job(), address=fake.address())
    session.add(user)


for _ in range(20):
    create_user()

session.commit()

We can see that we can use our models and instiate via constructor even though we have not specified any __init__, this is because the Base provides a default constructor that allows initialization from kwargs. And we are able to access the get or set the properties of the model to the actual types. This is made possible by the SQLAlchemy instrumentation which replaces all the columns properties by Python descriptors making it possible to assign directly the user name with user.name = "abc" and similarly allowing us to assign the profile to a UserProfile instance.

We then just add the user to the session with session.add. Notice that SQLAlchmey will know to add the UserProfile instance and we don’t have to explicitly add it to the session.

And finally we commit the changes to the session which will execute the SQL Query.

INFO sqlalchemy.engine.base.Engine ((21, 'Graphic designer', '275 Gregory Port\nNew Debbie, DC 71038'), (22, 'Web designer', '25805 Williams Glen Suite 382\nWest William, WV 68028'), (23, 'Administrator, local government', '80675 Brian Bridge\nPort Marcuschester, SD 59259'), (24, 'Lobbyist', '099 Shannon Creek Apt. 617\nAndersonland, IA 20753'), (25, 'Retail buyer', '06512 Garcia Roads Suite 504\nEast Maryfort, NM 74890'), (26, 'Teacher, primary school', '828 Torres Isle\nLarsonport, LA 83446'), (27, 'Call centre manager', '48376 Snyder Flats\nHensonburgh, MO 82689'), (28, 'Operational investment banker', '62107 Katherine Underpass\nTracyhaven, WA 07744')  ... displaying 10 of 20 total bound parameter sets ...  (39, 'Public relations account executive', '067 Scott Centers\nWest Mirandaport, VA 62169'), (40, 'Geophysicist/field seismologist', '90011 Schmidt Causeway Suite 313\nLake Ryanborough, FL 55528'))

Query

Now that we have inserted some users and profiles, we can look at how we would be able to run simple queries against our model. In order to run queries, we need to use the session which we defined earlier and build queries using the Query object.

users = session.query(User).all()

We query our model using .query() and specifying the selection. If we want to get all our users, we can specify query(User) or all our profile query(UserProfile). The output of query(...) is a query object, in order to execute the query, we need to force the iteration either with a loop or with a selector like .all, .one, .first or a slice operator which will be translated into a combination of skip/take. Returning a scalar can be done with .scalar which will return the first column of the row as a value.

session.query(User.name, UserProfile.address).filter(
    User.id == UserProfile.user_id
).first()

session.query(User, UserProfile.address).filter(
    User.id == UserProfile.user_id
).order_by(User.name)[2:5]

session.query(User.name).filter(User.id == 2).scalar()

We can also specify a mixture of selection, for example User.name, UserProfile.address will return a list of tuples str, str while User, UserProfile.address will return an instrumented User model with a str.

To apply filters to a query, we use filter. The most common operators are ==, !=, .like, .in_, and_ and or_.

from sqlalchemy import or_

session.query(UserProfile).filter(UserProfile.address.like("%Stravenue%")).all()

session.query(UserProfile).filter(
    or_(UserProfile.address.like("%Stravenue%"), UserProfile.job.like("%school%"))
).all()

session.query(UserProfile).filter(
    or_(UserProfile.address.like("%Stravenue%"), UserProfile.job.like("%school%"))
).all()

session.query(User).filter(User.name.in_(["Terry Franklin", "John Johnson"])).all()

We saw earlier that the quickest way to join table is to use an implicit query by selection both tables and using a filter on the identifier.

session.query(User, UserProfile).filter(
    User.id == UserProfile.user_id
)

This will create an implicit join and return the user and profile. If we don’t need to return the profile, we can instead use a SQL join:

session.query(User).join(UserProfile).filter(UserProfile.job.like("%school%")).all()

Lastly we mentioned during the creation of our model that the relationships were lazily loaded by default. This means that .profile will not be loaded until we access it resulting in a SELECT statement. In some cases we might want to earger load all profile. This can be achieved with .options(joinedload(...)).

from sqlalchemy.orm import joinedload

from sqlalchemy.orm import joinedload

users = session.query(User).options(joinedload(User.profile))
# Compare with: session.query(User)

for user in users:
    print(user.profile.job)

After executing the loop, we will see that we don’t incur extra SELECT statements. And that concludes today’s post!

Conclusion

Today we saw how we could get started with SQLAlchemy ORM. We started by looking at what an engine was, then moved on to define a simple database model which we then created on a in memory SQLite database. We then moved on to see how we could insert data, and lastly we looked at how we could query the data we inserted using the powerful Query API provided by SQLAlchemy. I hope you liked this post and I see you on the next one!