from sqlalchemy.connectors import Connector
+
import sys
import re
from sqlalchemy.sql import operators as sql_operators
from sqlalchemy.sql import functions as sql_functions
from sqlalchemy.sql import compiler
+from array import array as _array
from sqlalchemy.engine import base as engine_base, default
from sqlalchemy import types as sqltypes
return 'CAST(%s AS %s)' % (self.process(cast.clause), type_)
-
- def post_process_text(self, text):
- if '%%' in text:
- util.warn("The SQLAlchemy MySQLDB dialect now automatically escapes '%' in text() expressions to '%%'.")
- return text.replace('%', '%%')
-
def get_select_precolumns(self, select):
if isinstance(select._distinct, basestring):
return select._distinct.upper() + " "
raise
def do_begin_twophase(self, connection, xid):
- connection.execute("XA BEGIN %s", xid)
+ connection.execute(sql.text("XA BEGIN :xid"), xid=xid)
def do_prepare_twophase(self, connection, xid):
- connection.execute("XA END %s", xid)
- connection.execute("XA PREPARE %s", xid)
+ connection.execute(sql.text("XA END :xid"), xid=xid)
+ connection.execute(sql.text("XA PREPARE :xid"), xid=xid)
def do_rollback_twophase(self, connection, xid, is_prepared=True,
recover=False):
if not is_prepared:
- connection.execute("XA END %s", xid)
- connection.execute("XA ROLLBACK %s", xid)
+ connection.execute(sql.text("XA END :xid"), xid=xid)
+ connection.execute(sql.text("XA ROLLBACK :xid"), xid=xid)
def do_commit_twophase(self, connection, xid, is_prepared=True,
recover=False):
if not is_prepared:
self.do_prepare_twophase(connection, xid)
- connection.execute("XA COMMIT %s", xid)
+ connection.execute(sql.text("XA COMMIT :xid"), xid=xid)
def do_recover_twophase(self, connection):
resultset = connection.execute("XA RECOVER")
return False
def _compat_fetchall(self, rp, charset=None):
- return rp.fetchall()
+ """Proxy result rows to smooth over MySQL-Python driver inconsistencies."""
+
+ return [_DecodingRowProxy(row, charset) for row in rp.fetchall()]
def _compat_fetchone(self, rp, charset=None):
- return rp.fetchone()
+ """Proxy a result row to smooth over MySQL-Python driver inconsistencies."""
+
+ return _DecodingRowProxy(rp.fetchone(), charset)
def _extract_error_code(self, exception):
raise NotImplementedError()
table.metadata.tables[lc_alias] = table
def _detect_charset(self, connection):
- """Sniff out the character set in use for connection results."""
-
- # Allow user override, won't sniff if force_charset is set.
- if ('mysql', 'force_charset') in connection.info:
- return connection.info[('mysql', 'force_charset')]
-
- # Prefer 'character_set_results' for the current connection over the
- # value in the driver. SET NAMES or individual variable SETs will
- # change the charset without updating the driver's view of the world.
- #
- # If it's decided that issuing that sort of SQL leaves you SOL, then
- # this can prefer the driver value.
- rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'")
- opts = dict([(row[0], row[1]) for row in self._compat_fetchall(rs)])
-
- if 'character_set_results' in opts:
- return opts['character_set_results']
- # Still no charset on < 1.2.1 final...
- if 'character_set' in opts:
- return opts['character_set']
- else:
- util.warn(
- "Could not detect the connection character set. Assuming latin1.")
- return 'latin1'
- _detect_charset = engine_base.connection_memoize(
- ('mysql', 'charset'))(_detect_charset)
-
+ raise NotImplementedError()
def _detect_casing(self, connection):
"""Sniff out identifier case sensitivity.
log.class_logger(MySQLSchemaReflector)
+class _DecodingRowProxy(object):
+ """Return unicode-decoded values based on type inspection.
+
+ Smooth over data type issues (esp. with alpha driver versions) and
+ normalize strings as Unicode regardless of user-configured driver
+ encoding settings.
+
+ """
+
+ # Some MySQL-python versions can return some columns as
+ # sets.Set(['value']) (seriously) but thankfully that doesn't
+ # seem to come up in DDL queries.
+
+ def __init__(self, rowproxy, charset):
+ self.rowproxy = rowproxy
+ self.charset = charset
+ def __getitem__(self, index):
+ item = self.rowproxy[index]
+ if isinstance(item, _array):
+ item = item.tostring()
+ if self.charset and isinstance(item, str):
+ return item.decode(self.charset)
+ else:
+ return item
+ def __getattr__(self, attr):
+ item = getattr(self.rowproxy, attr)
+ if isinstance(item, _array):
+ item = item.tostring()
+ if self.charset and isinstance(item, str):
+ return item.decode(self.charset)
+ else:
+ return item
+
+
class _MySQLIdentifierPreparer(compiler.IdentifierPreparer):
"""MySQL-specific schema identifier configuration."""
create_engine('mysql:///mydb?charset=utf8&use_unicode=0')
"""
-from sqlalchemy.dialects.mysql.base import MySQLDialect, MySQLExecutionContext
+from sqlalchemy.dialects.mysql.base import MySQLDialect, MySQLExecutionContext, MySQLCompiler
from sqlalchemy.engine import base as engine_base, default
from sqlalchemy import exc, log, schema, sql, util
import re
-from array import array as _array
class MySQL_mysqldbExecutionContext(MySQLExecutionContext):
def _lastrowid(self, cursor):
return cursor.lastrowid
+class MySQL_mysqldbCompiler(MySQLCompiler):
+ def post_process_text(self, text):
+ if '%%' in text:
+ util.warn("The SQLAlchemy mysql+mysqldb dialect now automatically escapes '%' in text() expressions to '%%'.")
+ return text.replace('%', '%%')
+
class MySQL_mysqldb(MySQLDialect):
driver = 'mysqldb'
supports_unicode_statements = False
default_paramstyle = 'format'
execution_ctx_cls = MySQL_mysqldbExecutionContext
+ sql_compiler = MySQL_mysqldbCompiler
@classmethod
def dbapi(cls):
def _extract_error_code(self, exception):
return exception.orig.args[0]
+ @engine_base.connection_memoize(('mysql', 'charset'))
def _detect_charset(self, connection):
"""Sniff out the character set in use for connection results."""
"combination of MySQL server and MySQL-python. "
"MySQL-python >= 1.2.2 is recommended. Assuming latin1.")
return 'latin1'
- _detect_charset = engine_base.connection_memoize(
- ('mysql', 'charset'))(_detect_charset)
-
-
- def _compat_fetchall(self, rp, charset=None):
- """Proxy result rows to smooth over MySQL-Python driver inconsistencies."""
-
- return [_MySQLPythonRowProxy(row, charset) for row in rp.fetchall()]
-
- def _compat_fetchone(self, rp, charset=None):
- """Proxy a result row to smooth over MySQL-Python driver inconsistencies."""
-
- return _MySQLPythonRowProxy(rp.fetchone(), charset)
-
-class _MySQLPythonRowProxy(object):
- """Return consistent column values for all versions of MySQL-python.
-
- Smooth over data type issues (esp. with alpha driver versions) and
- normalize strings as Unicode regardless of user-configured driver
- encoding settings.
- """
-
- # Some MySQL-python versions can return some columns as
- # sets.Set(['value']) (seriously) but thankfully that doesn't
- # seem to come up in DDL queries.
-
- def __init__(self, rowproxy, charset):
- self.rowproxy = rowproxy
- self.charset = charset
- def __getitem__(self, index):
- item = self.rowproxy[index]
- if isinstance(item, _array):
- item = item.tostring()
- if self.charset and isinstance(item, str):
- return item.decode(self.charset)
- else:
- return item
- def __getattr__(self, attr):
- item = getattr(self.rowproxy, attr)
- if isinstance(item, _array):
- item = item.tostring()
- if self.charset and isinstance(item, str):
- return item.decode(self.charset)
- else:
- return item
dialect = MySQL_mysqldb
\ No newline at end of file
from sqlalchemy.dialects.mysql.base import MySQLDialect, MySQLExecutionContext
from sqlalchemy.connectors.pyodbc import PyODBCConnector
+from sqlalchemy.engine import base as engine_base
+from sqlalchemy import util
import re
class MySQL_pyodbcExecutionContext(MySQLExecutionContext):
def __init__(self, **kw):
MySQLDialect.__init__(self, **kw)
PyODBCConnector.__init__(self, **kw)
+
+ @engine_base.connection_memoize(('mysql', 'charset'))
+ def _detect_charset(self, connection):
+ """Sniff out the character set in use for connection results."""
+
+ # Allow user override, won't sniff if force_charset is set.
+ if ('mysql', 'force_charset') in connection.info:
+ return connection.info[('mysql', 'force_charset')]
+
+ # Prefer 'character_set_results' for the current connection over the
+ # value in the driver. SET NAMES or individual variable SETs will
+ # change the charset without updating the driver's view of the world.
+ #
+ # If it's decided that issuing that sort of SQL leaves you SOL, then
+ # this can prefer the driver value.
+ rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'")
+ opts = dict([(row[0], row[1]) for row in self._compat_fetchall(rs)])
+ for key in ('character_set_connection', 'character_set'):
+ if opts.get(key, None):
+ return opts[key]
+
+ util.warn("Could not detect the connection character set. Assuming latin1.")
+ return 'latin1'
def _extract_error_code(self, exception):
m = re.compile(r"\((\d+)\)").search(str(exception.orig.args))
def tearDownAll(self):
metadata.drop_all()
- @testing.fails_on_everything_except('firebird', 'maxdb', 'sqlite')
+ @testing.fails_on_everything_except('firebird', 'maxdb', 'sqlite', 'mysql+pyodbc')
def test_raw_qmark(self):
for conn in (testing.db, testing.db.connect()):
conn.execute("insert into users (user_id, user_name) values (?, ?)", (1,"jack"))
assert res.fetchall() == [(1, "jack"), (2, "fred"), (3, "ed"), (4, "horse"), (5, "barney"), (6, "donkey"), (7, 'sally')]
conn.execute("delete from users")
- @testing.fails_on_everything_except('mysql', 'postgres')
+ @testing.fails_on_everything_except('mysql+mysqldb', 'postgres')
# some psycopg2 versions bomb this.
def test_raw_sprintf(self):
for conn in (testing.db, testing.db.connect()):
# pyformat is supported for mysql, but skipping because a few driver
# versions have a bug that bombs out on this test. (1.2.2b3, 1.2.2c1, 1.2.2)
- @testing.skip_if(lambda: testing.against('mysql'), 'db-api flaky')
+ @testing.skip_if(lambda: testing.against('mysql+mysqldb'), 'db-api flaky')
@testing.fails_on_everything_except('postgres')
def test_raw_python(self):
for conn in (testing.db, testing.db.connect()):
def test_explicit_default_schema(self):
engine = testing.db
- if testing.against('mysql'):
+ if testing.against('mysql+mysqldb'):
schema = testing.db.url.database
elif testing.against('postgres'):
schema = 'public'
class Engineer(Person):
__mapper_args__ = {'polymorphic_identity':'engineer'}
- primary_language_id = Column(String(50), ForeignKey('languages.id'))
+ primary_language_id = Column(Integer, ForeignKey('languages.id'))
primary_language = relation("Language")
class Language(Base, ComparableEntity):
self.assert_(isinstance(x['unicode_text'], unicode) and x['unicode_text'] == unicodedata)
if isinstance(x['plain_varchar'], unicode):
# SQLLite and MSSQL return non-unicode data as unicode
- self.assert_(testing.against('sqlite', 'mssql'))
+ self.assert_(testing.against('sqlite', '+pyodbc'))
if not testing.against('sqlite'):
self.assert_(x['plain_varchar'] == unicodedata)
print "it's %s!" % testing.db.name
def decorate(fn):
fn_name = fn.__name__
def maybe(*args, **kw):
- if spec(db):
+ if spec(config.db):
msg = "'%s' unsupported on DB implementation '%s+%s': %s" % (
fn_name, config.db.name, config.db.driver, reason)
print msg
for query in queries:
if isinstance(query, basestring):
- if config.db.name == query:
+ if db_spec(query)(config.db):
return True
else:
name, op, spec = query
- if config.db.name != name:
+ if not db_spec(name)(config.db):
continue
have = config.db.dialect.server_version_info(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
