# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
+from sqlalchemy.dialects.sqlite import base as sqlite
+from sqlalchemy.dialects.postgres import base as postgres
+from sqlalchemy.dialects.mysql import base as mysql
+
__all__ = (
'access',
'informix',
'maxdb',
'mssql',
+ 'mysql',
+ 'postgres',
+ 'sqlite',
'oracle',
'sybase',
)
only the collation of character data.
"""
- self.__ddl_values = values
+ self._ddl_values = values
strip_values = []
for a in values:
else:
return self._extend_numeric(type_, 'REAL')
-
def visit_FLOAT(self, type_):
if self._mysql_type(type_) and type_.scale is not None and type_.precision is not None:
return self._extend_numeric(type_, "FLOAT(%s, %s)" % (type_.precision, type_.scale))
else:
return self.visit_BLOB(type_)
+ def visit_binary(self, type_):
+ return self.visit_BLOB(type_)
+
def visit_BINARY(self, type_):
if type_.length:
return "BINARY(%d)" % type_.length
return self._extend_string(type_, "ENUM(%s)" % ",".join(quoted_enums))
def visit_SET(self, type_):
- return self._extend_string("SET(%s)" % ",".join(type_._ddl_values))
+ return self._extend_string(type_, "SET(%s)" % ",".join(type_._ddl_values))
- def visit_BOOL(self, type):
+ def visit_BOOLEAN(self, type):
return "BOOL"
if rs:
rs.close()
+ @engine_base.connection_memoize(('mysql', 'server_version_info'))
def server_version_info(self, connection):
"""A tuple of the database server version.
cached per-Connection.
"""
+ # TODO: do we need to bypass ConnectionFairy here? other calls
+ # to this seem to not do that.
return self._server_version_info(connection.connection.connection)
- server_version_info = engine_base.connection_memoize(
- ('mysql', 'server_version_info'))(server_version_info)
def reflecttable(self, connection, table, include_columns):
"""Load column definitions from the server."""
# ANSI_QUOTES doesn't affect SHOW CREATE TABLE on < 4.1
preparer = MySQLIdentifierPreparer(self)
- self.reflector = reflector = MySQLSchemaReflector(self)
+ self.reflector = reflector = MySQLSchemaReflector(self, preparer)
sql = self._show_create_table(connection, table, charset)
if sql.startswith('CREATE ALGORITHM'):
class MySQLSchemaReflector(object):
"""Parses SHOW CREATE TABLE output."""
- def __init__(self, dialect):
+ def __init__(self, dialect, preparer=None):
"""Construct a MySQLSchemaReflector.
identifier_preparer
"""
self.dialect = dialect
- self.preparer = dialect.identifier_preparer
+ self.preparer = preparer or dialect.identifier_preparer
self._prep_regexes()
def reflect(self, connection, table, show_create, charset, only=None):
supports_pk_autoincrement = False
supports_default_values = True
supports_empty_insert = False
+ default_paramstyle = 'pyformat'
statement_compiler = PGCompiler
ddl_compiler = PGDDLCompiler
supports_default_values = True
supports_empty_insert = False
supports_cast = True
+ default_paramstyle = 'qmark'
statement_compiler = SQLiteCompiler
ddl_compiler = SQLiteDDLCompiler
type_compiler = SQLiteTypeCompiler
within a URL.
"""
-import sqlalchemy.databases
+# not sure what this was used for
+#import sqlalchemy.databases
+
from sqlalchemy.engine.base import (
BufferedColumnResultProxy,
BufferedColumnRow,
table_args.append(Column('c%s' % index, type_(*args, **kw)))
numeric_table = Table(*table_args)
- gen = testing.db.dialect.schemagenerator(testing.db.dialect, testing.db, None, None)
+ gen = testing.db.dialect.ddl_compiler(testing.db.dialect, numeric_table)
for col in numeric_table.c:
index = int(col.name[1:])
table_args.append(Column('c%s' % index, type_(*args, **kw)))
charset_table = Table(*table_args)
- gen = testing.db.dialect.schemagenerator(testing.db.dialect, testing.db, None, None)
+ gen = testing.db.dialect.ddl_compiler(testing.db.dialect, charset_table)
for col in charset_table.c:
index = int(col.name[1:])
for table in tables:
for i, reflected in enumerate(table.c):
- assert isinstance(reflected.type, type(expected[i]))
+ assert isinstance(reflected.type, type(expected[i])), \
+ "element %d: %r not instance of %r" % (i, reflected.type, type(expected[i]))
finally:
db.execute('DROP VIEW mysql_types_v')
finally:
class RawReflectionTest(TestBase):
def setUp(self):
self.dialect = mysql.dialect()
- self.reflector = mysql.MySQLSchemaReflector(
- self.dialect.identifier_preparer)
+ self.reflector = mysql.MySQLSchemaReflector(self.dialect)
def test_key_reflection(self):
regex = self.reflector._re_key
def colspec(c):
- return testing.db.dialect.schemagenerator(testing.db.dialect,
- testing.db, None, None).get_column_specification(c)
+ return testing.db.dialect.ddl_compiler(testing.db.dialect, c.table).get_column_specification(c)
if __name__ == "__main__":
testenv.main()
from sqlalchemy.sql import table, column, label, compiler
from sqlalchemy.sql.expression import ClauseList
from sqlalchemy.engine import default
-from sqlalchemy.databases import mysql, oracle, firebird, mssql
-from sqlalchemy.dialects.sqlite import pysqlite as sqlite
-from sqlalchemy.dialects.postgres import psycopg2 as postgres
+from sqlalchemy.databases import *
from testlib import *
table1 = table('mytable',
s1 = select([table1.c.myid, table1.c.myid.label('foobar'), func.hoho(table1.c.name), func.lala(table1.c.name).label('gg')])
assert s1.c.keys() == ['myid', 'foobar', 'hoho(mytable.name)', 'gg']
- from sqlalchemy.databases.sqlite import SLNumeric
meta = MetaData()
t1 = Table('mytable', meta, Column('col1', Integer))
(table1.c.name, 'name', 'mytable.name', None),
(table1.c.myid==12, 'mytable.myid = :myid_1', 'mytable.myid = :myid_1', 'anon_1'),
(func.hoho(table1.c.myid), 'hoho(mytable.myid)', 'hoho(mytable.myid)', 'hoho_1'),
- (cast(table1.c.name, SLNumeric), 'CAST(mytable.name AS NUMERIC(10, 2))', 'CAST(mytable.name AS NUMERIC(10, 2))', 'anon_1'),
+ (cast(table1.c.name, sqlite.SLNumeric), 'CAST(mytable.name AS NUMERIC(10, 2))', 'CAST(mytable.name AS NUMERIC(10, 2))', 'anon_1'),
(t1.c.col1, 'col1', 'mytable.col1', None),
(column('some wacky thing'), 'some wacky thing', '"some wacky thing"', '')
):
from sqlalchemy.sql import operators
from testlib.testing import eq_
import sqlalchemy.engine.url as url
-from sqlalchemy.databases import mssql, oracle, mysql, firebird
-from sqlalchemy.dialects.sqlite import pysqlite as sqlite
-from sqlalchemy.dialects.postgres import psycopg2 as postgres
+from sqlalchemy.databases import *
from testlib import *
def all_dialects():
import sqlalchemy.databases as d
for name in d.__all__:
- mod = getattr(__import__('sqlalchemy.databases.%s' % name).databases, name)
- yield mod.dialect()
- import sqlalchemy.dialects as d
- for name in d.__all__:
- mod = getattr(__import__('sqlalchemy.dialects.%s.base' % name).dialects, name).base
+ # TEMPORARY
+ mod = getattr(d, name, None)
+ if not mod:
+ mod = getattr(__import__('sqlalchemy.databases.%s' % name).databases, name)
yield mod.dialect()
class ReconnectFixture(object):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
