add #286: add parent_table in utils

sqlalchemy_continuum/manager.py

@@ -129,6 +129,7 @@ def reset(self):
         self.tables = {}
         self.pending_classes = []
         self.association_tables = set()
+        self.association_tables_map = {}
         self.association_version_tables = set()
         self.declarative_base = None
         self.version_class_map = {}

sqlalchemy_continuum/relationship_builder.py

@@ -337,7 +337,7 @@ def build_association_version_tables(self):
             # may have already been created if we visiting the 'other' side of
             # a self-referential many-to-many relationship
             self.association_version_table = metadata.tables[table_name]
-
+        self.manager.association_tables_map[self.association_version_table] = column.table
     def __call__(self):
         """
         Builds reflected relationship between version classes based on given

sqlalchemy_continuum/utils.py

@@ -90,6 +90,22 @@ def parent_class(version_cls):
     return get_versioning_manager(version_cls).parent_class_map[version_cls]
 
 
+def parent_table(table):
+    versioning_manager = get_versioning_manager(table)
+    if table in versioning_manager.association_version_tables:
+        return versioning_manager.association_tables_map.get(table, None)
+
+    parent_table = next(
+        iter(
+            parent_class(m).__table__
+            for m in versioning_manager.parent_class_map
+            if m.__table__ == table
+        ),
+        None,
+    )
+    return parent_table
+
+
 def transaction_class(cls):
     """
     Return the associated transaction class for given versioned SQLAlchemy

tests/utils/test_parent_table.py

@@ -0,0 +1,41 @@
+import copy
+import datetime
+import sqlalchemy as sa
+from sqlalchemy_continuum.utils import parent_table, version_table
+
+from tests import TestCase
+
+
+class TestParentTable(TestCase):
+
+    def create_models(self):
+        super().create_models()
+
+        article_author_table = sa.Table(
+            'article_author',
+            self.Model.metadata,
+            sa.Column('article_id', sa.Integer, sa.ForeignKey('article.id'), primary_key=True, nullable=False),
+            sa.Column('author_id', sa.Integer, sa.ForeignKey('author.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 Author(self.Model):
+            __tablename__ = 'author'
+            __versioned__ = {
+                'baseclass': (self.Model, )
+            }
+            id = sa.Column(sa.Integer, autoincrement=True, primary_key=True)
+            name = sa.Column(sa.Unicode(255))
+            articles = sa.orm.relationship('Article', secondary=article_author_table, backref='author')
+
+
+        self.Author = Author
+        self.article_author_table = article_author_table
+
+    def test_parent_table_from_version_table(self):
+        author_version_table = version_table(self.Author.__table__)
+        assert parent_table(author_version_table) == self.Author.__table__
+
+    def test_parent_table_from_association_table(self):
+        versioned_article_author_table = version_table(self.article_author_table)
+        assert parent_table(versioned_article_author_table) == self.article_author_table