'CHAR': sqltypes.CHAR,
'DATE': sqltypes.DATE,
'DATETIME': sqltypes.DATETIME,
+ 'DOUBLE': sqltypes.FLOAT,
'DECIMAL': sqltypes.DECIMAL,
'FLOAT': sqltypes.FLOAT,
'INT': sqltypes.INTEGER,
return columns
def _get_column_info(self, name, type_, nullable, default, primary_key):
- match = re.match(r'(\w+)(\(.*?\))?', type_)
+ match = re.match(r'([\w ]+)(\(.*?\))?', type_)
if match:
coltype = match.group(1)
args = match.group(2)
else:
- coltype = "VARCHAR"
+ coltype = ''
args = ''
- try:
- coltype = self.ischema_names[coltype]
- if args is not None:
- args = re.findall(r'(\d+)', args)
- coltype = coltype(*[int(a) for a in args])
- except KeyError:
- util.warn("Did not recognize type '%s' of column '%s'" %
- (coltype, name))
- coltype = sqltypes.NullType()
+ coltype = self._resolve_col_affinity(coltype)
+ if args is not None:
+ args = re.findall(r'(\d+)', args)
+ coltype = coltype(*[int(a) for a in args])
if default is not None:
default = util.text_type(default)
'primary_key': primary_key,
}
+ def _resolve_col_affinity(self, coltype):
+ """Return a data type from a reflected column, using affinity tules.
+
+ SQLite's goal for universal compatability introduces some complexity
+ during reflection, as a column's defined type might not actually be a
+ type that SQLite understands - or indeed, my not be defined *at all*.
+ Internally, SQLite handles this with a 'data type affinity' for each
+ column definition, mapping to one of 'TEXT', 'NUMERIC', 'INTEGER',
+ 'REAL', or 'NONE' (raw bits). The algorithm that determines this is
+ listed in http://www.sqlite.org/datatype3.html section 2.1.
+
+ This method allows SQLAlchemy to support that algorithm, while still
+ providing access to smarter reflection utilities by regcognizing
+ column definitions that SQLite only supports through affinity (like
+ DATE and DOUBLE).
+ """
+ if coltype in self.ischema_names:
+ return self.ischema_names[coltype]
+ if 'INT' in coltype:
+ return sqltypes.INTEGER
+ elif 'CHAR' in coltype or 'CLOB' in coltype or 'TEXT' in coltype:
+ return sqltypes.TEXT,
+ elif 'BLOB' in coltype or not coltype:
+ return sqltypes.NullType
+ elif 'REAL' in coltype or 'FLOA' in coltype or 'DOUB' in coltype:
+ return sqltype.REAL
+ else:
+ return sqltypes.NUMERIC
+
@reflection.cache
def get_pk_constraint(self, connection, table_name, schema=None, **kw):
cols = self.get_columns(connection, table_name, schema, **kw)
#!coding: utf-8
"""SQLite-specific tests."""
+import os
+from collections import Counter
+import datetime
from sqlalchemy.testing import eq_, assert_raises, \
assert_raises_message
-import datetime
from sqlalchemy import Table, String, select, Text, CHAR, bindparam, Column,\
Unicode, Date, MetaData, UnicodeText, Time, Integer, TIMESTAMP, \
Boolean, func, NUMERIC, DateTime, extract, ForeignKey, text, Numeric,\
from sqlalchemy.testing import fixtures, AssertsCompiledSQL, \
AssertsExecutionResults, engines
from sqlalchemy import testing
-import os
from sqlalchemy.schema import CreateTable
class TestTypes(fixtures.TestBase, AssertsExecutionResults):
"""Test that the boolean only treats 1 as True
"""
-
+
meta = MetaData(testing.db)
t = Table('bool_table', meta, Column('id', Integer,
primary_key=True), Column('boo',
)
t.create()
t2 = Table('t', MetaData(), autoload=True, autoload_with=testing.db)
- assert isinstance(t2.c.x.type, sqltypes.NullType)
- assert isinstance(t2.c.y.type, sqltypes.NullType)
+ assert isinstance(t2.c.x.type, sqltypes.Numeric)
+ assert isinstance(t2.c.y.type, sqltypes.Numeric)
class DateTimeTest(fixtures.TestBase, AssertsCompiledSQL):
set([con.name for con in c.constraints]),
set([None, None])
)
+
+
+class ColumnTypeAffinityReflectionTest(fixtures.TestBase):
+ """Tests on data type affinities for SQLite during relection.
+
+ See http://www.sqlite.org/datatype3.html - section 2.
+ """
+ __only_on__ = 'sqlite'
+
+ def setup(self):
+ testing.db.execute("""
+ CREATE TABLE a (
+ "id" INTEGER PRIMARY KEY,
+ "foo" DOUBLE,
+ "bar" DECIMAL(19,4),
+ "baz" VARCHAR(200),
+ "boff",
+ "biff" LONGTEXT
+ )""")
+ # These example names come from section 2.2 of the datatype docs,
+ # after pruning out types which we convert to more convenient types
+ self.example_typenames_integer = ["TINYINT", "MEDIUMINT", "INT2",
+ "UNSIGNED BIG INT", "INT8"]
+ self.example_typenames_text = ["CHARACTER(20)", "CLOB",
+ "VARYING CHARACTER(70)", "NATIVE CHARACTER(70)"]
+ self.example_typenames_none = [""]
+ self.example_typenames_real = ["DOUBLE PRECISION"]
+ cols = ["i%d %s" % (n, t) for n, t in enumerate(
+ self.example_typenames_integer)]
+ cols += ["t%d %s" % (n, t) for n, t in enumerate(
+ self.example_typenames_text)]
+ cols += ["o%d %s" % (n, t) for n, t in enumerate(
+ self.example_typenames_none)]
+ cols += ["n%d %s" % (n, t) for n, t in enumerate(
+ self.example_typenames_real)]
+ cols = ','.join(cols)
+ testing.db.execute("CREATE TABLE b (%s)" % (cols,))
+
+ def teardown(self):
+ testing.db.execute("drop table a")
+ testing.db.execute("drop table b")
+
+ def test_can_reflect_with_affinity(self):
+ "Test that 'affinity-types' don't break reflection outright."
+ meta = MetaData()
+ a = Table('a', meta, autoload=True, autoload_with=testing.db)
+ eq_(len(a.columns), 6)
+
+ def test_correct_reflection_with_affinity(self):
+ "Test that coltypes are detected correctly from affinity rules."
+ meta = MetaData()
+ b = Table('b', meta, autoload=True, autoload_with=testing.db)
+ typecounts = Counter(type(col.type) for col in b.columns)
+ eq_(typecounts[sqltypes.INTEGER], len(self.example_typenames_integer))
+ eq_(typecounts[sqltypes.TEXT], len(self.example_typenames_text))
+ eq_(typecounts[sqltypes.NullType], len(self.example_typenames_none))
+ eq_(typecounts[sqltypes.REAL], len(self.example_typenames_real))
import sqlalchemy as sa
from sqlalchemy import schema, events, event, inspect
from sqlalchemy import MetaData, Integer, String
-from sqlalchemy.testing import ComparesTables, \
- engines, AssertsCompiledSQL, fixtures
+from sqlalchemy.testing import (ComparesTables, engines, AssertsCompiledSQL,
+ fixtures, skip)
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
from sqlalchemy import testing
t2a = Table('test2', m2, autoload=True)
assert t2a._autoincrement_column is t2a.c.id2
+ @skip('sqlite')
@testing.provide_metadata
def test_unknown_types(self):
+ """Test the handling of unknown types for the given dialect.
+
+ sqlite is skipped because it has special rules for unknown types using
+ 'affinity types' - this feature is tested in that dialect's test spec.
+ """
meta = self.metadata
t = Table("test", meta,
Column('foo', sa.DateTime))
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
