arguments and should return a true value for any table to reflect.
"""
- reflect_opts = {'autoload': True}
if bind is None:
bind = _bind_or_error(self)
- conn = None
+
+ if bind.engine is not bind:
+ conn = bind
+ close = False
else:
- reflect_opts['autoload_with'] = bind
conn = bind.contextual_connect()
+ close = True
+
+ reflect_opts = {
+ 'autoload': True,
+ 'autoload_with': bind
+ }
if schema is None:
schema = self.schema
connection=conn))
if views:
available.update(
- bind.dialect.get_view_names(conn or bind, schema)
+ bind.dialect.get_view_names(conn, schema)
)
current = set(self.tables.iterkeys())
for name in load:
Table(name, self, **reflect_opts)
finally:
- if conn is not None and \
- conn is not bind:
+ if close:
conn.close()
def append_ddl_listener(self, event_name, listener):
table_b2 = Table('false', meta2, autoload=True)
table_c2 = Table('is', meta2, autoload=True)
+ @testing.provide_metadata
+ def _test_reflect_uses_bind(self, fn):
+ from sqlalchemy.pool import AssertionPool
+ e = engines.testing_engine(options={"poolclass": AssertionPool})
+ fn(e)
+
+ def test_reflect_uses_bind_constructor_conn(self):
+ self._test_reflect_uses_bind(lambda e: MetaData(e.connect(),
+ reflect=True))
+
+ def test_reflect_uses_bind_constructor_engine(self):
+ self._test_reflect_uses_bind(lambda e: MetaData(e, reflect=True))
+
+ def test_reflect_uses_bind_constructor_conn_reflect(self):
+ self._test_reflect_uses_bind(lambda e: MetaData(e.connect()).reflect())
+
+ def test_reflect_uses_bind_constructor_engine_reflect(self):
+ self._test_reflect_uses_bind(lambda e: MetaData(e).reflect())
+
+ def test_reflect_uses_bind_conn_reflect(self):
+ self._test_reflect_uses_bind(lambda e: MetaData().reflect(e.connect()))
+
+ def test_reflect_uses_bind_engine_reflect(self):
+ self._test_reflect_uses_bind(lambda e: MetaData().reflect(e))
+
@testing.provide_metadata
def test_reflect_all(self):
existing = testing.db.table_names()