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?
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.