fix(39): set end_transaction_id for association columns as well

since we are in direction to support versioning control on tables
as well as ORM models, extended support for filling end transaction
id column in association table as well when manager has validity
strategy enabled.
this for now is a kind of hacky as we don't support specifying
configurations on table level so we refer configurations mentioned
in manager and apply it globally, alternatively we can give an
option to disable this behaviour even though validity strategy is
enabled but didn't encorporated it because it seemed bit of an
over-configgyy solution for now want to let the value be set.

sqlalchemy_history/unit_of_work.py

@@ -13,6 +13,7 @@
     tx_column_name,
     versioned_column_properties,
 )
+from sqlalchemy_history.schema import update_end_tx_column
 
 
 class UnitOfWork(object):
@@ -249,6 +250,18 @@ def create_association_versions(self, session):
                 **{self.manager.options["transaction_column_name"]: self.current_transaction.id}
             )
             session.execute(stmt)
+            if self.manager.options["strategy"] == "validity":
+                # FIXME: Currently we don't support setting versioning behaviour on bare table level
+                #        so we only refer to global configuration of manager and if it is set to validity
+                #        we always assign value to end_transaction_id table. which should have no impact as
+                #        if user is not using this column it won't impact except executing a redundant
+                #        assocition query for end_transaction_id, maybe provide a flag to disable this?
+                update_end_tx_column(
+                    stmt.table,
+                    end_tx_column_name=self.manager.options["end_transaction_column_name"],
+                    tx_column_name=self.manager.options["transaction_column_name"],
+                    conn=session.connection(),
+                )
         self.pending_statements = []
 
     def make_versions(self, session):

tests/schema/test_update_end_transaction_id.py

@@ -1,11 +1,48 @@
+import datetime
+import sqlalchemy as sa
+
 from sqlalchemy_history import version_class
+from sqlalchemy_history.utils import version_table
 from sqlalchemy_history.schema import update_end_tx_column
-from tests import TestCase
+from sqlalchemy_history.operation import Operation
+from tests import TestCase, create_test_cases
 
 
-class TestSchemaTools(TestCase):
+class UpdateEndTransactionID(TestCase):
     versioning_strategy = "validity"
 
+    def create_models(self):
+        super().create_models()
+        article_label_table = sa.Table(
+            "article_label",
+            self.Model.metadata,
+            sa.Column(
+                "article_id", sa.Integer, sa.ForeignKey("article.id"), primary_key=True, nullable=False
+            ),
+            sa.Column("label_id", sa.Integer, sa.ForeignKey("label.id"), primary_key=True, nullable=False),
+            sa.Column(
+                "created_date",
+                sa.DateTime,
+                nullable=False,
+                server_default=sa.func.current_timestamp(),
+                default=datetime.datetime.utcnow,
+            ),
+        )
+
+        class Label(self.Model):
+            __tablename__ = "label"
+            __versioned__ = {}
+
+            id = sa.Column(
+                sa.Integer, sa.Sequence(f"{__tablename__}_seq"), autoincrement=True, primary_key=True
+            )
+            name = sa.Column(sa.Unicode(255))
+            article_id = sa.Column(sa.Integer, sa.ForeignKey(self.Article.id))
+            article = sa.orm.relationship(self.Article, backref="labels", secondary=article_label_table)
+
+        self.article_label_table = article_label_table
+        self.Label = Label
+
     def _insert(self, values):
         table = version_class(self.Article).__table__
         stmt = table.insert().values(values)
@@ -53,16 +90,75 @@ def test_update_end_transaction_id(self):
                 "operation_type": 2,
             }
         )
+        if self.versioning_strategy == "validity":
+            update_end_tx_column(table, conn=self.session)
+            rows = self.session.execute("SELECT * FROM article_version ORDER BY transaction_id").fetchall()
+            assert rows[0].transaction_id == 1
+            assert rows[0].end_transaction_id == 2
+            assert rows[1].transaction_id == 2
+            assert rows[1].end_transaction_id == 4
+            assert rows[2].transaction_id == 3
+            assert rows[2].end_transaction_id == 5
+            assert rows[3].transaction_id == 4
+            assert rows[3].end_transaction_id is None
+            assert rows[4].transaction_id == 5
+            assert rows[4].end_transaction_id is None
+        elif self.versioning_strategy == "subquery":
+            rows = self.session.execute("SELECT * FROM article_version ORDER BY transaction_id").fetchall()
+            assert not hasattr(rows[0], "end_transaction_id")
+
+    def test_assoc_update_end_transaction_id(self):
+        article_label_table_version = version_table(self.article_label_table)
+
+        label = self.Label(name="first label")
+        label2 = self.Label(name="second label")
+        article = self.Article(name="Some article", content="Some content", labels=[label])
+        self.session.add(article)
+        self.session.commit()
+        article.labels = [label2]
+        self.session.commit()
+        article.labels = [label, label2]
+        self.session.commit()
+
+        article.labels = [label]
+        self.session.commit()
+        rows = (
+            self.session.query(article_label_table_version)
+            .order_by(article_label_table_version.c.transaction_id)
+            .all()
+        )
+        if self.versioning_strategy == "validity":
+            assert rows[0].label_id == label.id
+            assert rows[0].transaction_id == 1
+            assert rows[0].end_transaction_id == 2
+            assert rows[0].operation_type == Operation.INSERT
+
+            assert rows[1].label_id == label.id
+            assert rows[1].transaction_id == 2
+            assert rows[1].end_transaction_id == 3
+            assert rows[1].operation_type == Operation.DELETE
+
+            assert {rows[2].label_id, rows[3].label_id} == {label.id, label2.id}
+            assert rows[2].transaction_id == 2
+            assert rows[2].end_transaction_id == 4
+            assert rows[2].operation_type == Operation.INSERT
+            assert rows[3].transaction_id == 3
+            assert rows[3].end_transaction_id is None
+            assert rows[3].operation_type == Operation.INSERT
+
+            assert rows[4].label_id == label2.id
+            assert rows[4].transaction_id == 4
+            assert rows[4].end_transaction_id is None
+            assert rows[4].operation_type == Operation.DELETE
+        elif self.versioning_strategy == "subquery":
+            assert not hasattr(rows[0], "end_transaction_id")
+
+
+setting_variants = {
+    "versioning_strategy": [
+        "subquery",
+        "validity",
+    ],
+}
 
-        update_end_tx_column(table, conn=self.session)
-        rows = self.session.execute("SELECT * FROM article_version ORDER BY transaction_id").fetchall()
-        assert rows[0].transaction_id == 1
-        assert rows[0].end_transaction_id == 2
-        assert rows[1].transaction_id == 2
-        assert rows[1].end_transaction_id == 4
-        assert rows[2].transaction_id == 3
-        assert rows[2].end_transaction_id == 5
-        assert rows[3].transaction_id == 4
-        assert rows[3].end_transaction_id is None
-        assert rows[4].transaction_id == 5
-        assert rows[4].end_transaction_id is None
+create_test_cases(UpdateEndTransactionID, setting_variants)