Handle column default values

Author: rundef

README.md

@@ -27,6 +27,8 @@ Unlike traditional in-memory solutions like SQLite, `sqlalchemy-memory` fully av
 
 It is also perfect for **applications that need a lightweight, high-performance store** compatible with SQLAlchemy, such as backtesting engines, simulators, or other tools where you don't want to maintain a separate in-memory layer alongside your database models.
 
+Data is kept purely in RAM and is **volatile**: it is **not persisted across application restarts** and is **cleared when the engine is disposed**.
+
 ## Features
 
 - **SQLAlchemy 2.0 support**: ORM & Core expressions, sync & async modes

sqlalchemy_memory/base/store.py

@@ -1,4 +1,7 @@
 from collections import defaultdict
+from sqlalchemy import func
+from sqlalchemy.sql.elements import TextClause
+from datetime import datetime
 
 from ..logger import logger
 
@@ -80,6 +83,8 @@ def commit(self):
                 if pk_value in self.data_by_pk[tablename].keys():
                     raise Exception(f"Cannot have duplicate PK value {pk_value} for table '{tablename}'")
 
+                self._apply_column_defaults(obj)
+
                 logger.debug(f"Adding {obj} to table '{tablename}'")
 
                 self.data[tablename].append(obj)
@@ -155,3 +160,37 @@ def _assign_primary_key_if_needed(self, obj):
             self._pk_counter[table] = max(self._pk_counter[table], current_id)
 
         return current_id
+
+    def _apply_column_defaults(self, obj):
+        """
+        Apply default and server_default values to an ORM object.
+        """
+
+        for column in obj.__table__.columns:
+            attr_name = column.name
+            current_value = getattr(obj, attr_name, None)
+
+            if current_value is not None:
+                continue
+
+            elif column.default is not None:
+                if callable(column.default.arg):
+                    try:
+                        value = column.default.arg()
+                    except TypeError:
+                        value = column.default.arg(ctx=None)
+                else:
+                    value = column.default.arg
+
+                setattr(obj, attr_name, value)
+
+            elif column.server_default is not None:
+                if isinstance(column.server_default.arg, TextClause):
+                    text_value = column.server_default.arg.text
+                    setattr(obj, attr_name, text_value)
+
+                elif isinstance(column.server_default.arg, func.now().__class__):
+                    setattr(obj, attr_name, datetime.utcnow())
+
+                else:
+                    raise Exception(f"Unhandled server_default type: {type(column.server_default)}")

tests/models.py

@@ -1,5 +1,6 @@
 from sqlalchemy.orm import declarative_base, mapped_column, Mapped
-from sqlalchemy import JSON
+from sqlalchemy import JSON, func, text
+from datetime import datetime
 
 Base = declarative_base()
 
@@ -17,8 +18,13 @@ class Product(Base):
     id: Mapped[int] = mapped_column(primary_key=True)
     active: Mapped[bool] = mapped_column(default=True, index=True)
     name: Mapped[str] = mapped_column(nullable=False)
-    category: Mapped[str] = mapped_column(nullable=False, index=True)
+    category: Mapped[str] = mapped_column(index=True, server_default=text("unknown"))
     data: Mapped[dict] = mapped_column(JSON)
 
+    created_at: Mapped[datetime] = mapped_column(
+        server_default=func.now(),
+        nullable=False
+    )
+
     def __repr__(self):
         return f"Product(id={self.id} name={self.name})"

tests/test_advanced.py

@@ -1,4 +1,5 @@
 from sqlalchemy import select, func
+from datetime import datetime, date
 import pytest
 
 from models import Item, Product
@@ -117,6 +118,91 @@ def test_json_extract_filter(self, SessionFactory, pattern, value, expected_ids)
             results = session.execute(stmt).scalars().all()
             assert {item.id for item in results} == expected_ids
 
+    def test_default_values(self, SessionFactory):
+        dt = datetime(2025, 1, 1, 2, 3, 4)
+
+        with SessionFactory() as session:
+            session.add_all([
+                Product(id=5, name="foo", category="A"),
+                Product(name="bar", active=False, created_at=dt),
+            ])
+            session.commit()
+
+            products = session.execute(select(Product)).scalars().all()
+
+            assert products[0].active
+            assert products[0].created_at is not None
+            assert products[0].category == "A"
+
+            assert products[1].id == 6
+            assert not products[1].active
+            assert products[1].created_at == dt
+            assert products[1].category == "unknown"
+
+    @pytest.mark.parametrize(
+        "operator, value, expected_ids",
+        [
+            ("is", True, {1, 3}),
+            ("is_not", True, {2}),
+            ("is", False, {2}),
+            ("is_not", False, {1, 3}),
+        ]
+    )
+    def test_is_filter(self, SessionFactory, operator, value, expected_ids):
+        with SessionFactory() as session:
+            session.add_all([
+                Product(id=1, name="foo", active=True),
+                Product(id=2, name="bar", active=False),
+                Product(id=3, name="foobar", active=True),
+            ])
+            session.commit()
+
+            stmt = select(Product)
+            if operator == "is":
+                stmt = stmt.where(Product.active.is_(value))
+            else:
+                stmt = stmt.where(Product.active.is_not(value))
+
+            results = session.execute(stmt).scalars().all()
+            assert {item.id for item in results} == expected_ids
+
+    @pytest.mark.parametrize(
+        "operator, value, expected_ids",
+        [
+            ("==", date(2025, 1, 1), {1}),
+            ("!=", date(2025, 1, 2), {1, 3, 4}),
+            (">", date(2025, 1, 2), {3, 4}),
+            (">", date(2025, 1, 10), set()),
+        ]
+    )
+    def test_date_filter(self, SessionFactory, operator, value, expected_ids):
+        """
+        stmt = select(NetLiquidationValue).where(
+                func.DATE(NetLiquidationValue.created_at) == _date
+            )
+
+        """
+
+        with SessionFactory() as session:
+            session.add_all([
+                Product(id=1, name="foo", created_at=datetime(2025, 1, 1, 1, 1, 1)),
+                Product(id=2, name="bar", created_at=datetime(2025, 1, 2, 2, 2, 2)),
+                Product(id=3, name="foobar", created_at=datetime(2025, 1, 3, 3, 3, 3)),
+                Product(id=4, name="barfoo", created_at=datetime(2025, 1, 4, 4, 4, 4)),
+            ])
+            session.commit()
+
+            stmt = select(Product)
+            if operator == "==":
+                stmt = stmt.where(func.DATE(Product.created_at) == value)
+            elif operator == "!=":
+                stmt = stmt.where(func.DATE(Product.created_at) != value)
+            elif operator == ">":
+                stmt = stmt.where(func.DATE(Product.created_at) > value)
+
+            results = session.execute(stmt).scalars().all()
+            assert {item.id for item in results} == expected_ids
+
     def test_session_inception(self, SessionFactory):
         with SessionFactory() as session1:
             session1.add(Item(id=1, name="foo"))