ORM

Expanse's database component is powered by SQLAlchemy and, as such, comes with a powerful ORM. The ORM helps you map your database tables to Python classes and provides a way to interact with your database in a more intuitive way.

Creating models

To create a new model, you can use the make model command. The command accepts the name of the model as an argument.

./beam make model Product

This will create a file product.py in the app/models directory. The model will look like this:

product.py
from expanse.database.orm.model import Model


class Product(Model):
    __tablename__ = 'products'

By default, Expanse will assume that the table name is the plural, snake-case form of the model name. If you want to use a different table name, you can give the --table option to the make model command.

./beam make model Product --table special_products

Customizing models

The models generated by Expanse do nothing by default. You will need to define the columns and relationships. This documentation will not cover all the features of SQLAlchemy, so you can refer to the SQLAlchemy documentation for more information, but one thing of note is that Expanse makes use of the Declarative Dataclass Mapping feature of SQLAlchemy. This means that your models will be standard dataclasses, making them leaner and more intuitive.

Let's add a few columns to our Product model:

product.py
from expanse.database.orm import column
from sqlalchemy import BigInteger
from sqlalchemy.orm import Mapped

from expanse.database.orm.model import Model


class Product(Model):
    __tablename__ = 'products'

    id: Mapped[int] = column(BigInteger(), primary_key=True, autoincrement=True)
    sku: Mapped[str] = column(nullable=False, unique=True)

The column function is a helper function that makes it easier to define columns. It acts as a way to declare column attributes as well as the field() function of the dataclasses module. This is means you can declare any dataclass specific field attributes directly in column.

product.py
from expanse.database.orm import column
from sqlalchemy import BigInteger
from sqlalchemy.orm import Mapped

from expanse.database.orm.model import Model


class Product(Model):
    __tablename__ = 'products'

    id: Mapped[int] = column(BigInteger(), primary_key=True, autoincrement=True)
    sku: Mapped[str] = column(nullable=False, unique=True)
    details: Mapped[str | None] = column(default=None)

Reusing column declarations

If you find yourself repeating the same column declarations in multiple models — for instance primary keys — you can annotate a column declaration to make it reusable.

product.py
from typing import Annotated

from expanse.database.orm import column
from sqlalchemy import BigInteger, Identity
from sqlalchemy.dialects import sqlite
from sqlalchemy.orm import Mapped

from expanse.database.orm.model import Model

primary_key = Annotated[
    int,
    column(
        BigInteger().with_variant(sqlite.INTEGER(), "sqlite"),
        Identity(always=True),
        primary_key=True,
    ),
]


class Product(Model):
    __tablename__ = 'products'

    id: Mapped[primary_key] = column()
    sku: Mapped[str] = column(nullable=False, unique=True)
    details: Mapped[str | None] = column(default=None)

Due to a limitation in how annotated types work, you will need to give a default value to the column by using the column() function without any arguments.

After you have done modifications to your model, do not forget to create the corresponding migration either manually or automatically.

Retrieving models

To retrieve models from the database, you can use a Session (or AsyncSession) instance.

from typing import Sequence

from expanse.database.session import Session
from expanse.http.response import Response
from sqlalchemy import select

from app.models.product import Product


class ProductController:
    def index(self, session: Session) -> Response:
        products: Sequence[Product] = session.scalars(select(Product)).all()

        ...
from typing import Sequence

from expanse.database.session import AsyncSession
from expanse.http.response import Response
from sqlalchemy import select

from app.models.product import Product


class ProductController:
    async def index(self, session: AsyncSession) -> Response:
        products: Sequence[Product] = (await session.scalars(select(Product))).all()

        ...

Retrieving a single model

If you want to retrieve a single model, you can use the scalar method.

from expanse.database.session import Session
from expanse.http.response import Response
from sqlalchemy import select

from app.models.product import Product


class ProductController:
    def show(self, session: Session, sku: str) -> Response:
        product: Product = session.scalar(
            select(Product).where(Product.sku == sku)
        )

        ...

scalar() will return a single instance of the model or None if no results are found.

If you prefer to raise an exception when no results are found, you can use the scalars().one() method.

from expanse.database.session import Session
from expanse.http.response import Response
from sqlalchemy import select

from app.models.product import Product


class ProductController:
    def show(self, session: Session, sku: str) -> Response:
        product: Product = session.scalars(
            select(Product).where(Product.sku == sku)
        ).one()

        ...

This will raise a NoResultFound exception if no results matching the query were found or a MultipleResultsFound exception if there were multiple results matching the query.

NoResultFound exceptions are automatically caught by Expanse and will return a 404 Not Found response automatically, unless they are caught explicitly by your application.

Inserting and updating models

Inserting new models

To insert new models into your database, you will need to instantiate them, add them to the Session instance and then commit the session. If you don't commit the session, the changes will not be saved to the database.

from expanse.common.http.json import JSON
from expanse.database.session import Session
from expanse.http.response import Response
from pydantic import BaseModel

from app.models.product import Product


class ProductData(BaseModel):
    sku: str
    details: str | None


class ProductController:
    def create(self, session: Session, data: JSON[ProductData]) -> Response:
        product = Product()

        product.sku = data.sku
        product.details = data.details

        session.add(product)

        session.commit()

        ...

In this example, we retrieve the request data as a validated instance of ProductData and then create a new instance and set its attributes. We then add the instance to the session and commit the session, so that the model is saved.

Mass assignment

What if you would like to instantiate the model with all the attributes at once? Let's try it.

from expanse.common.http.json import JSON
from expanse.database.session import Session
from expanse.http.response import Response
from pydantic import BaseModel

from app.models.product import Product


class ProductData(BaseModel):
    sku: str
    details: str | None


class ProductController:
    def create(self, session: Session, data: JSON[ProductData]) -> Response:
        product = Product(**data.model_dump())

        session.add(product)

        session.commit()

        ...

If you call the corresponding endpoint, you should get an error: TypeError: __init__() got an unexpected keyword argument 'sku'.

This is expected as models in Expanse are, by default, protected against mass assignment. You must explicitly specify which column can be mass-assigned by using init=True in the column() function in your model.

product.py
class Product(Model):
    __tablename__ = 'products'

    id: Mapped[primary_key] = column()
    sku: Mapped[str] = column(nullable=False, unique=True, init=True)
    details: Mapped[str | None] = column(default=None, init=True)

Now, if you call the endpoint again, the model will be created successfully.

Updating existing models

Updating existing models is as simple as retrieving the model from the database, modifying its attributes and then committing the session.

from expanse.common.http.json import JSON
from expanse.database.session import Session
from expanse.http.response import Response
from pydantic import BaseModel
from sqlalchemy import select

from app.models.product import Product


class ProductData(BaseModel):
    details: str | None


class ProductController:
    def update(
        self, session: Session, sku: str, data: JSON[ProductData]
    ) -> Response:
        product: Product = session.scalars(
            select(Product).where(Product.sku == sku)
        ).one()

        product.details = data.details

        session.commit()

        ...