SQLAlchemy ORM: translate between None in SQLAchemy ORM instance attributes to a specific database column value?

Question

SQLAlchemy ORM generally turns instance attributes that are None into NULL column values in the database (and vice-versa), but I'd like SQLAlchemy to convert between None values in the ORM and a specific value in the database (e.g. MyTable(latitude=None) is tied to mytable.latitude=-999.9 in the database).

Is there a way get SQLAlchemy to translate back and forth between None in ORM models and specific values in the database?

Answer 1

You can achieve this in SQLAlchemy with event listeners. In particular, with Mapper Events and Session Events.

I will use these event listeners: before_insert, before_update, loaded_as_persistent to demonstrate how to do it. There might be other related events you will need to modify.

Here's the example:

from sqlalchemy import create_engine, event, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

engine = create_engine(
    "postgresql+psycopg2://postgres:pw2023@localhost:5432/qa",
)

Session = sessionmaker(
    bind=engine,
)


class Base(DeclarativeBase):
    pass


class Test(Base):
    __tablename__ = "test"

    id: Mapped[int] = mapped_column(primary_key=True)
    data: Mapped[float] = mapped_column()


@event.listens_for(Test, 'before_insert')
def my_before_insert_listener(mapper, connection, target):
    if target.data is None:
        print("before_insert triggered.")
        target.data = -999.9


@event.listens_for(Test, 'before_update')
def receive_before_update(mapper, connection, target):
    if target.data is None:
        print("before_update triggered.")
        target.data = -999.9


@event.listens_for(Session, 'loaded_as_persistent')
def receive_before_attach(session, instance):
    if type(instance) == Test and instance.data == -999.9:
        print("loaded_as_persistent triggered.")
        instance.data = None


if __name__ == "__main__":
    Base.metadata.create_all(engine)

    # saving data, one with data=None, one with data=100
    print("# Saving initial data:")
    with Session() as session:
        session.add(Test(id=1))
        session.add(Test(id=2, data=100))
        session.commit()

    print("# Reading saved values:")
    with Session() as session:
        test1 = session.get(Test, 1)
        print(f"A field value of -999.9 is rendered as {test1.data}.")
        test2 = session.get(Test, 2)
        print(f"A field value of 100 is rendered as {test2.data}.")

    print("# Updating field value of 100 to None:")
    with Session() as session:
        test2 = session.get(Test, 2)
        test2.data = None
        session.commit()

    print("# Reading the updated value.")
    with Session() as session:
        test2 = session.scalar(select(Test).filter_by(id=2))
        print(f"A field value of -999.9 is rendered as: {test2.data}")

Sample output:

# Saving initial data:
before_insert triggered.

# Reading saved values:
loaded_as_persistent triggered.
A field value of -999.9 is rendered as None.
A field value of 100 is rendered as 100.0.

# Updating field value of 100 to None:
before_update triggered.

# Reading the updated value.
loaded_as_persistent triggered.
A field value of -999.9 is rendered as: None

Table values from the database:

|id |data  |
|---|------|
|1  |-999.9|
|2  |-999.9|

PS: You might want to just use getter/setter methods.