Test SQLAlchemy standalone without creating Flask app

SQLAlchemy is quite popular ORM in Python commonly used with Flask applications. SQLAlchemy has even a special version Flask-SQLAlchemy.

There is many tutorials on using SQLAlchemy in context of Flask application. However sometimes we need only to verify whether our data model created using SQLAlchemy ORM works as it’s expected and don’t want to create a Flask app for this purpose. Here I will show how to test a simple data model without creating a Flask app. We will use extremely simple scripts with few lines of code to migrate existing data model to database and perform simple queries.

First install required libraries:

sudo apt-get update
sudo apt-get install postgresql postgresql-contrib libpq-dev
pip install psycopg2 SQLAlchemy

Let’s create configuration file config.py:

# Scheme: "postgres+psycopg2://<USERNAME>:<PASSWORD>@<IP_ADDRESS>:<PORT>/<DATABASE_NAME>"
DATABASE_URI = "postgres+psycopg2://postgres:123@localhost/test"

Here we specify DATABASE_URI constant for connection to our database.

Then create data model in models.py:

from sqlalchemy.ext.declarative import declarative_base                                                                                         
from sqlalchemy import Column, Float, Integer, String
Base = declarative_base()
class Products(database.Base):
    __tablename__ = 'products'
    id = Column(UUID(as_uuid=True),
        primary_key=True,
        default=uuid.uuid4,
        index=True
    )
    title = Column(String)
    subtitle = Column(String)
    price = Column(Float)
    rating = Column(Float)
    description = Column(String)

We specify type UUID for Id column to be unique. Id can also be Integer date type. primary_key=True means that this column is Primay Key (PK).

Now create test script crud.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from config import DATABASE_URI
from models import Base
                                                                                                                                                
engine = create_engine(DATABASE_URI)
Base.metadata.create_all(engine)

Here we will first create table for our single model Products.

Now we can test creating, updating and deleting some entities.

Create a session object:

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

Import models:

from models import Products

Let’s create one record

product = Products(
    title="trainers",
    subtitle='Trainers In White',
    price=76,
    rating=4.6,
    discount=None,
    description="Sneakers (also known as athletic shoes, tennis shoes,gym shoes, runners, takkies, or trainers) are shoes primarily designed for sports or other forms of physical exercise, but which are now also often used for everyday wear.",
    updatedAt=func.now()
)
s.add(product)
s.commit()

We can check if our product was saved. List products

products = app.session.query(Products).all()

Then we can search for particular product by Id

products = app.session.query(Products).filter_by(id=product_id).first()

We can update particular product

product.title = data.get('title', None)
product.img = data.get('img', None)
product.price = data.get('price', None)
app.session.add(product)
app.session.commit()

and remove products

app.session.delete(product)
app.session.commit()

We can wrap session object by context manager for more convenient using:

@contextmanager
def session_scope():
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

and use it to get

with session_scope() as s:
    product = Products(
        title="trainers",
        ...
    )
    s.add(product)

With context manager we dont’ need to call commit() on session object thus less lines of code for procedure.