overhaul session.binds to support classes fully; fix up tests
Issue #3035
resolved
right now session.__binds uses mappers completely. This makes it impossible to feed in abstract classes and also because it links to base_mapper it's not possible to change the bind along a class hierarchy either (like a concrete hierarchy).
propose that __binds stores classes directly, and this can likely be a fallback in addition to what we have now. this is kind of a proof of concept
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 89d9946..f63adfb 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -636,13 +636,20 @@ class Session(_SessionClassMethods):
if binds is not None:
for mapperortable, bind in binds.items():
- insp = inspect(mapperortable)
- if insp.is_selectable:
- self.bind_table(mapperortable, bind)
- elif insp.is_mapper:
- self.bind_mapper(mapperortable, bind)
+ try:
+ insp = inspect(mapperortable)
+ except exc.NoInspectionAvailable:
+ if isinstance(mapperortable, type):
+ self.bind_cls(mapperortable, bind)
+ else:
+ raise exc.ArgumentError("Not acceptable bind target: %s" % mapperortable)
else:
- assert False
+ if insp.is_selectable:
+ self.bind_table(mapperortable, bind)
+ elif insp.is_mapper:
+ self.bind_mapper(mapperortable, bind)
+ else:
+ assert False
if not self.autocommit:
@@ -1039,6 +1046,17 @@ class Session(_SessionClassMethods):
for t in mapper._all_tables:
self.__binds[t] = bind
+ def bind_class(self, cls, bind):
+ """Bind operations for an unmapped class to a Connectable.
+
+ The behavior here is that a subclass of cls which is mapped
+ will be linked to this bind.
+
+ .. versionadded:: 1.0
+
+ """
+ self.__binds[cls] = bind
+
def bind_table(self, table, bind):
"""Bind operations on a Table to a Connectable.
@@ -1124,6 +1142,12 @@ class Session(_SessionClassMethods):
return self.__binds[c_mapper.base_mapper]
elif c_mapper.mapped_table in self.__binds:
return self.__binds[c_mapper.mapped_table]
+ else:
+ for cls in c_mapper.class_.__mro__:
+ if cls in self.__binds:
+ bind = self.__binds[cls]
+ self.bind_mapper(c_mapper, bind)
+ return bind
if clause is not None:
for t in sql_util.find_tables(clause, include_crud=True):
if t in self.__binds:
the tests here also need a rework, with an emphasis on straight unit testing of get_bind(). test_session -> BindTest is really just integration tests.
Comments (4)
-
reporter -
reporter Issue
#3202was marked as a duplicate of this issue. -
reporter additional test from
#3202:from sqlalchemy import Column, Integer, ForeignKey, create_engine, orm from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class A(Base): __tablename__ = 'A' id = Column(Integer, primary_key=True) type = Column(Integer, nullable=False) __mapper_args__ = {'polymorphic_on': type} class B(A): __tablename__ = 'B' id = Column(ForeignKey(A.id), primary_key=True) __mapper_args__ = {'polymorphic_identity': 1} engine = create_engine('sqlite://') binds = {table: engine for table in Base.metadata.sorted_tables} db = orm.sessionmaker(binds=binds)() print db.get_bind(A) # Engine(sqlite://) print db.get_bind(B) # sqlalchemy.exc.UnboundExecutionError: Could not locate a bind configured on mapper Mapper|B|B or this Session
-
reporter - changed status to resolved
- Improvements to the mechanism used by :class:
.Session
to locate "binds" (e.g. engines to use), such engines can be associated with mixin classes, concrete subclasses, as well as a wider variety of table metadata such as joined inheritance tables. fixes#3035
→ <<cset 6de1a878702b>>
- Log in to comment
here's a better patch, everything passes except for some exception raises: