From 8669eda82dcc84b726ab420106dfea86fc8a066f Mon Sep 17 00:00:00 2001 From: Michael Trier Date: Sun, 28 Dec 2008 21:07:57 +0000 Subject: [PATCH] Added in a new MSGenericBinary type. This maps to the Binary type so it can implement the specialized behavior of treating length specified types as fixed-width Binary types and non-length types as an unbound variable length Binary type. --- CHANGES | 6 +++ lib/sqlalchemy/databases/mssql.py | 29 ++++++++++-- test/dialect/mssql.py | 76 ++++++++++++++++++++++++++++++- test/sql/testtypes.py | 1 + 4 files changed, 106 insertions(+), 6 deletions(-) diff --git a/CHANGES b/CHANGES index 4f2f07f679..791833d6fa 100644 --- a/CHANGES +++ b/CHANGES @@ -43,7 +43,13 @@ CHANGES sent to connection.execute() and friends. [ticket:935] - mssql + - Added in a new MSGenericBinary type. This maps to the Binary + type so it can implement the specialized behavior of treating + length specified types as fixed-width Binary types and + non-length types as an unbound variable length Binary type. + - Added in new types: MSVarBinary and MSImage. [ticket:1249] + - Added in the MSReal and MSNText types. - bugfixes, behavioral changes diff --git a/lib/sqlalchemy/databases/mssql.py b/lib/sqlalchemy/databases/mssql.py index 0844745de0..4d2176f854 100644 --- a/lib/sqlalchemy/databases/mssql.py +++ b/lib/sqlalchemy/databases/mssql.py @@ -676,7 +676,23 @@ class MSNChar(_StringType, sqltypes.NCHAR): return self._extend("NCHAR") -class MSBinary(sqltypes.Binary): +class MSGenericBinary(sqltypes.Binary): + """The Binary type assumes that a Binary specification without a length + is an unbound Binary type whereas one with a length specification results + in a fixed length Binary type. + + If you want standard MSSQL ``BINARY`` behavior use the ``MSBinary`` type. + + """ + + def get_col_spec(self): + if self.length: + return "BINARY(%s)" % self.length + else: + return "IMAGE" + + +class MSBinary(MSGenericBinary): def get_col_spec(self): if self.length: return "BINARY(%s)" % self.length @@ -684,7 +700,7 @@ class MSBinary(sqltypes.Binary): return "BINARY" -class MSVarBinary(MSBinary): +class MSVarBinary(MSGenericBinary): def get_col_spec(self): if self.length: return "VARBINARY(%s)" % self.length @@ -692,7 +708,7 @@ class MSVarBinary(MSBinary): return "VARBINARY" -class MSImage(MSBinary): +class MSImage(MSGenericBinary): def get_col_spec(self): return "IMAGE" @@ -720,22 +736,27 @@ class MSBoolean(sqltypes.Boolean): return value and True or False return process + class MSTimeStamp(sqltypes.TIMESTAMP): def get_col_spec(self): return "TIMESTAMP" + class MSMoney(sqltypes.TypeEngine): def get_col_spec(self): return "MONEY" + class MSSmallMoney(MSMoney): def get_col_spec(self): return "SMALLMONEY" + class MSUniqueIdentifier(sqltypes.TypeEngine): def get_col_spec(self): return "UNIQUEIDENTIFIER" + class MSVariant(sqltypes.TypeEngine): def get_col_spec(self): return "SQL_VARIANT" @@ -851,7 +872,7 @@ class MSSQLDialect(default.DefaultDialect): sqltypes.Date : MSDate, sqltypes.Time : MSTime, sqltypes.String : MSString, - sqltypes.Binary : MSBinary, + sqltypes.Binary : MSGenericBinary, sqltypes.Boolean : MSBoolean, sqltypes.Text : MSText, sqltypes.UnicodeText : MSNText, diff --git a/test/dialect/mssql.py b/test/dialect/mssql.py index c4056fd9be..f63787e748 100755 --- a/test/dialect/mssql.py +++ b/test/dialect/mssql.py @@ -1,8 +1,8 @@ import testenv; testenv.configure_for_tests() -import re +import os, pickleable, re from sqlalchemy import * +from sqlalchemy import types, exc from sqlalchemy.orm import * -from sqlalchemy import exc from sqlalchemy.sql import table, column from sqlalchemy.databases import mssql import sqlalchemy.engine.url as url @@ -826,5 +826,77 @@ def colspec(c): testing.db, None, None).get_column_specification(c) +class BinaryTest(TestBase, AssertsExecutionResults): + """Test the Binary and VarBinary types""" + def setUpAll(self): + global binary_table, MyPickleType + + class MyPickleType(types.TypeDecorator): + impl = PickleType + + def process_bind_param(self, value, dialect): + if value: + value.stuff = 'this is modified stuff' + return value + + def process_result_value(self, value, dialect): + if value: + value.stuff = 'this is the right stuff' + return value + + binary_table = Table('binary_table', MetaData(testing.db), + Column('primary_id', Integer, Sequence('binary_id_seq', optional=True), primary_key=True), + Column('data', mssql.MSVarBinary(8000)), + Column('data_image', mssql.MSImage), + Column('data_slice', Binary(100)), + Column('misc', String(30)), + # construct PickleType with non-native pickle module, since cPickle uses relative module + # loading and confuses this test's parent package 'sql' with the 'sqlalchemy.sql' package relative + # to the 'types' module + Column('pickled', PickleType), + Column('mypickle', MyPickleType) + ) + binary_table.create() + + def tearDown(self): + binary_table.delete().execute() + + def tearDownAll(self): + binary_table.drop() + + def testbinary(self): + testobj1 = pickleable.Foo('im foo 1') + testobj2 = pickleable.Foo('im foo 2') + testobj3 = pickleable.Foo('im foo 3') + + stream1 =self.load_stream('binary_data_one.dat') + stream2 =self.load_stream('binary_data_two.dat') + binary_table.insert().execute(primary_id=1, misc='binary_data_one.dat', data=stream1, data_image=stream1, data_slice=stream1[0:100], pickled=testobj1, mypickle=testobj3) + binary_table.insert().execute(primary_id=2, misc='binary_data_two.dat', data=stream2, data_image=stream2, data_slice=stream2[0:99], pickled=testobj2) + binary_table.insert().execute(primary_id=3, misc='binary_data_two.dat', data=None, data_image=None, data_slice=stream2[0:99], pickled=None) + + for stmt in ( + binary_table.select(order_by=binary_table.c.primary_id), + text("select * from binary_table order by binary_table.primary_id", typemap={'pickled':PickleType, 'mypickle':MyPickleType}, bind=testing.db) + ): + l = stmt.execute().fetchall() + self.assertEquals(list(stream1), list(l[0]['data'])) + + paddedstream = list(stream1[0:100]) + paddedstream.extend(['\x00'] * (100 - len(paddedstream))) + self.assertEquals(paddedstream, list(l[0]['data_slice'])) + + self.assertEquals(list(stream2), list(l[1]['data'])) + self.assertEquals(list(stream2), list(l[1]['data_image'])) + self.assertEquals(testobj1, l[0]['pickled']) + self.assertEquals(testobj2, l[1]['pickled']) + self.assertEquals(testobj3.moredata, l[0]['mypickle'].moredata) + self.assertEquals(l[0]['mypickle'].stuff, 'this is the right stuff') + + def load_stream(self, name, len=3000): + f = os.path.join(os.path.dirname(testenv.__file__), name) + return file(f).read(len) + + if __name__ == "__main__": testenv.main() diff --git a/test/sql/testtypes.py b/test/sql/testtypes.py index d7fa5987c6..ef4a355d3c 100644 --- a/test/sql/testtypes.py +++ b/test/sql/testtypes.py @@ -433,6 +433,7 @@ class BinaryTest(TestBase, AssertsExecutionResults): def tearDownAll(self): binary_table.drop() + @testing.fails_on('mssql', 'MSSQl BINARY type right pads the fixed length with \x00') def testbinary(self): testobj1 = pickleable.Foo('im foo 1') testobj2 = pickleable.Foo('im foo 2') -- 2.47.3