secondary_update.append(associationrow)
if secondary_delete:
- secondary_delete.sort()
# TODO: precompile the delete/insert queries?
statement = self.secondary.delete(sql.and_(*[c == sql.bindparam(c.key, type_=c.type) for c in self.secondary.c if c.key in associationrow]))
result = connection.execute(statement, secondary_delete)
from collections import deque
import re
from sqlalchemy import util
+import operator
__all__ = ['VisitableType', 'Visitable', 'ClauseVisitor',
'CloningVisitor', 'ReplacingCloningVisitor', 'iterate',
# for use by the compiler
visit_name = cls.__dict__["__visit_name__"]
if isinstance(visit_name, str):
- func_text = "def _compiler_dispatch(self, visitor, **kw):\n"\
- " return visitor.visit_%s(self, **kw)" % visit_name
+ getter = operator.attrgetter("visit_%s" % visit_name)
+ def _compiler_dispatch(self, visitor, **kw):
+ return getter(visitor)(self, **kw)
else:
- func_text = "def _compiler_dispatch(self, visitor, **kw):\n"\
- " return getattr(visitor, 'visit_%s' % self.__visit_name__)(self, **kw)"
+ def _compiler_dispatch(self, visitor, **kw):
+ return getattr(visitor, 'visit_%s' % self.__visit_name__)(self, **kw)
- env = locals().copy()
- exec func_text in env
- cls._compiler_dispatch = env['_compiler_dispatch']
+ cls._compiler_dispatch = _compiler_dispatch
super(VisitableType, cls).__init__(clsname, bases, dict)
def test_profile_2_insert(self):
self.test_baseline_2_insert()
- @profiling.function_call_count(6385)
+ @profiling.function_call_count(6783)
def test_profile_3_properties(self):
self.test_baseline_3_properties()
- @profiling.function_call_count(22508)
+ @profiling.function_call_count(23861)
def test_profile_4_expressions(self):
self.test_baseline_4_expressions()