Intro

SQLAlchemy is a very popular ORM Python library.

Key Concepts

Engine

See Engine Configuration. The engine is the starting point for SQLAlchemy application. The very basic usage is to create an engine from a postgresql url:

from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2://scott:tiger@localhost:5432/mydatabase")

Session

See Session Basics. A session is a “holding zone” for the orm-mapped objects and the database. we can configure properties like autocommit and autoflush, and these settings can impact the performance of the application.

We can use a sessionmaker to provide a factory for Session objects with a fixed configuration:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/")
Session = sessionmaker(engine)

Declarative Base

See Declarative Mapping Styles. You can think of the “declarative base” as the base class which can be extended to create your own mapping class.

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)

Unit Tests

To set up unit tests for SQLAlchemy, first we set up an in-memory databas:

@pytest.fixture(scope="module")
def db():
    engine = create_engine("sqlite:///:memory:")
    TestSession = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base.metadata.create_all(engine)
    db = TestSession()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(engine)

Then we can treat it as a normal tests:

def test_create_user_in_db(db):
    user = User(
        username="testuser",
        email="testuser@example.com"
    )

    db.add(user)
    db.commit()
    db.refresh(user)

    assert user.id is not None
    assert user.username == "testuser"
    assert user.email == "testuser@example.com"

def test_get_user_by_username(db):
    user = db.query(User).filter(User.username == "testuser").first()

    assert user is not None
    assert user.username == "testuser"
    assert user.email == "testuser@example.com"