def do_release_savepoint(self, connection, name):
pass
- @base.connection_memoize(('dialect', 'default_schema_name'))
def get_default_schema_name(self, connection):
query = "SELECT user_name() as user_name;"
user_name = connection.scalar(sql.text(query))
def _extract_error_code(self, exception):
raise NotImplementedError()
- @engine_base.connection_memoize(('dialect', 'default_schema_name'))
def get_default_schema_name(self, connection):
return connection.execute('SELECT DATABASE()').scalar()
lc_alias = schema._get_table_key(table.name, table.schema)
table.metadata.tables[lc_alias] = table
- #@engine_base.connection_memoize(('mysql', 'charset'))
def _detect_charset(self, connection):
raise NotImplementedError()
- #@engine_base.connection_memoize(('mysql', 'lower_case_table_names'))
def _detect_casing(self, connection):
"""Sniff out identifier case sensitivity.
row.close()
return cs
- #@engine_base.connection_memoize(('mysql', 'collations'))
def _detect_collations(self, connection):
"""Pull the active COLLATIONS list from the server.
def do_ping(self, connection):
connection.ping()
- #@engine_base.connection_memoize(('mysql', 'server_version_info'))
def _get_server_version_info(self, connection):
dbapi_con = connection.connection
version = []
else:
return name.encode(self.encoding)
- @base.connection_memoize(('dialect', 'default_schema_name'))
def get_default_schema_name(self, connection):
return self._normalize_name(connection.execute('SELECT USER FROM DUAL').scalar())
resultset = connection.execute(sql.text("SELECT gid FROM pg_prepared_xacts"))
return [row[0] for row in resultset]
- @base.connection_memoize(('dialect', 'default_schema_name'))
def get_default_schema_name(self, connection):
return connection.scalar("select current_schema()")
cx = engine.connect()
meta = MetaData()
-
- assert ('mysql', 'charset') not in cx.info
- assert ('mysql', 'force_charset') not in cx.info
-
- cx.execute(text("SELECT 1")).fetchall()
- assert ('mysql', 'charset') not in cx.info
-
- meta.reflect(cx)
- assert ('mysql', 'charset') in cx.info
-
- cx.execute(text("SET @squiznart=123"))
- assert ('mysql', 'charset') in cx.info
-
- # the charset invalidation is very conservative
- cx.execute(text("SET TIMESTAMP = DEFAULT"))
- assert ('mysql', 'charset') not in cx.info
-
- cx.info[('mysql', 'force_charset')] = 'latin1'
-
assert engine.dialect._detect_charset(cx) == 'latin1'
- assert cx.info[('mysql', 'charset')] == 'latin1'
-
- del cx.info[('mysql', 'force_charset')]
- del cx.info[('mysql', 'charset')]
meta.reflect(cx)
- assert ('mysql', 'charset') in cx.info
-
- # String execution doesn't go through the detector.
- cx.execute("SET TIMESTAMP = DEFAULT")
- assert ('mysql', 'charset') in cx.info
+ assert cx.dialect._connection_charset == 'latin1'
class MatchTest(TestBase, AssertsCompiledSQL):