Mike Bayer avatar Mike Bayer committed 5ef6d29

- got firebird running
- add some failure cases
- [bug] Firebird now uses strict "ansi bind rules"
so that bound parameters don't render in the
columns clause of a statement - they render
literally instead.

- [bug] Support for passing datetime as date when
using the DateTime type with Firebird; other
dialects support this.

Comments (0)

Files changed (9)

     no length is attempted to be emitted, same
     way as MySQL. [ticket:2505]
 
+  - [bug] Firebird now uses strict "ansi bind rules"
+    so that bound parameters don't render in the
+    columns clause of a statement - they render
+    literally instead.
+
+  - [bug] Support for passing datetime as date when
+    using the DateTime type with Firebird; other
+    dialects support this.
+
 - mysql
   - [bug] Dialect no longer emits expensive server
     collations query, as well as server casing,

lib/sqlalchemy/dialects/firebird/base.py

 class _StringType(sqltypes.String):
     """Base for Firebird string types."""
 
-    def __init__(self, charset = None, **kw):
+    def __init__(self, charset=None, **kw):
         self.charset = charset
         super(_StringType, self).__init__(**kw)
 
     """Firebird VARCHAR type"""
     __visit_name__ = 'VARCHAR'
 
-    def __init__(self, length = None, **kwargs):
+    def __init__(self, length=None, **kwargs):
         super(VARCHAR, self).__init__(length=length, **kwargs)
 
 class CHAR(_StringType, sqltypes.CHAR):
     """Firebird CHAR type"""
     __visit_name__ = 'CHAR'
 
-    def __init__(self, length = None, **kwargs):
+    def __init__(self, length=None, **kwargs):
         super(CHAR, self).__init__(length=length, **kwargs)
 
+
+class _FBDateTime(sqltypes.DateTime):
+    def bind_processor(self, dialect):
+        def process(value):
+            if type(value) == datetime.date:
+                return datetime.datetime(value.year, value.month, value.day)
+            else:
+                return value
+        return process
+
 colspecs = {
+    sqltypes.DateTime: _FBDateTime
 }
 
 ischema_names = {
 class FBCompiler(sql.compiler.SQLCompiler):
     """Firebird specific idiosyncrasies"""
 
+    ansi_bind_rules = True
+
     #def visit_contains_op_binary(self, binary, operator, **kw):
         # cant use CONTAINING b.c. it's case insensitive.
 
     #def visit_notcontains_op_binary(self, binary, operator, **kw):
         # cant use NOT CONTAINING b.c. it's case insensitive.
 
+    def visit_now_func(self, fn, **kw):
+        return "CURRENT_TIMESTAMP"
+
     def visit_startswith_op_binary(self, binary, operator, **kw):
         return '%s STARTING WITH %s' % (
                             binary.left._compiler_dispatch(self, **kw),
 
     visit_char_length_func = visit_length_func
 
-    def function_argspec(self, func, **kw):
+    def _function_argspec(self, func, **kw):
         # TODO: this probably will need to be
         # narrowed to a fixed list, some no-arg functions
         # may require parens - see similar example in the oracle

test/engine/test_execute.py

     @testing.fails_on("postgresql+pg8000",
             "pg8000 still doesn't allow single % without params")
     def test_no_params_option(self):
-        stmt = "SELECT '%'"
-        if testing.against('oracle'):
-            stmt += " FROM DUAL"
+        stmt = "SELECT '%'" + testing.db.dialect.statement_compiler(
+                                    testing.db.dialect, None).default_from()
+
         conn = testing.db.connect()
         result = conn.\
                 execution_options(no_parameters=True).\
                     ('INSERT INTO t1 (c1, c2)', {
                         'c2': 'some data', 'c1': 5},
                         (5, 'some data')),
-                    ('SELECT lower', {'lower_2': 'Foo'}, 
+                    ('SELECT lower', {'lower_2': 'Foo'},
                         ('Foo', )),
                     ('INSERT INTO t1 (c1, c2)',
                      {'c2': 'foo', 'c1': 6},
                     ('CREATE TABLE t1', {}, ()),
                     ('INSERT INTO t1 (c1, c2)', {'c2': 'some data', 'c1'
                      : 5}, (5, 'some data')),
-                    ('SELECT lower', {'lower_2': 'Foo'}, 
+                    ('SELECT lower', {'lower_2': 'Foo'},
                         ('Foo', )),
                     ('INSERT INTO t1 (c1, c2)', {'c2': 'foo', 'c1': 6},
                      (6, 'foo')),

test/lib/requires.py

     return _chain_decorators_on(
         fn,
         skip_if(lambda: testing.against('oracle'),
+            "non-standard SELECT scalar syntax"),
+        skip_if(lambda: testing.against('firebird'),
             "non-standard SELECT scalar syntax")
     )
 

test/orm/test_froms.py

     @testing.fails_on('postgresql+zxjdbc',
                       "zxjdbc parses the SQL itself before passing on "
                       "to PG, doesn't parse this")
+    @testing.fails_on("firebird", "unknown")
     def test_values_with_boolean_selects(self):
         """Tests a values clause that works with select boolean
         evaluations"""
             eq_(results, [(User(name='jack'), 'jack')])
         self.assert_sql_count(testing.db, go, 1)
 
+    @testing.fails_on("firebird", "unknown")
     @testing.fails_on('postgresql+pg8000', "'type oid 705 not mapped to py type' (due to literal)")
     def test_self_referential(self):
         Order = self.classes.Order

test/orm/test_query.py

         assert u.addresses[0].email_address == 'jack@bean.com'
         assert u.orders[1].items[2].description == 'item 5'
 
-    @testing.fails_on_everything_except('sqlite', '+pyodbc', '+zxjdbc', 'mysql+oursql')
-    def test_query_str(self):
-        User = self.classes.User
-
-        s = create_session()
-        q = s.query(User).filter(User.id==1)
-        eq_(
-            str(q).replace('\n',''),
-            'SELECT users.id AS users_id, users.name AS users_name FROM users WHERE users.id = ?'
-            )
 
 class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
     def test_no_limit_offset(self):
         )
 
 
-    @testing.fails_on('mysql', "mysql doesn't support intersect")
+    @testing.requires.intersect
     def test_intersect(self):
         User = self.classes.User
 

test/sql/test_case_statement.py

 
         assert_raises(exc.ArgumentError, case, [("x", "y")])
 
-        self.assert_compile(case([("x", "y")], value=t.c.col1), "CASE test.col1 WHEN :param_1 THEN :param_2 END")
-        self.assert_compile(case([(t.c.col1==7, "y")], else_="z"), "CASE WHEN (test.col1 = :col1_1) THEN :param_1 ELSE :param_2 END")
+        self.assert_compile(case([("x", "y")], value=t.c.col1),
+                "CASE test.col1 WHEN :param_1 THEN :param_2 END")
+        self.assert_compile(case([(t.c.col1 == 7, "y")], else_="z"),
+                "CASE WHEN (test.col1 = :col1_1) THEN :param_1 ELSE :param_2 END")
 
     def test_text_doesnt_explode(self):
 
                   ))]).order_by(info_table.c.info),
 
         ]:
-            eq_(s.execute().fetchall(), [
-                (u'no', ), (u'no', ), (u'no', ), (u'yes', ),
-                (u'no', ), (u'no', ),
-                ])
+            if testing.against("firebird"):
+                eq_(s.execute().fetchall(), [
+                    ('no ', ), ('no ', ), ('no ', ), ('yes', ),
+                    ('no ', ), ('no ', ),
+                    ])
+            else:
+                eq_(s.execute().fetchall(), [
+                    ('no', ), ('no', ), ('no', ), ('yes', ),
+                    ('no', ), ('no', ),
+                    ])
 
 
 

test/sql/test_query.py

             l.append(row)
         self.assert_(len(l) == 3)
 
-    @testing.fails_on('firebird', "kinterbasdb doesn't send full type information")
     @testing.requires.subqueries
     def test_anonymous_rows(self):
         users.insert().execute(
                               use_labels=labels),
                  [(3, 'a'), (2, 'b'), (1, None)])
 
-    @testing.fails_on('mssql+pyodbc', 
+    @testing.fails_on('mssql+pyodbc',
         "pyodbc result row doesn't support slicing")
     def test_column_slices(self):
         users.insert().execute(user_id=1, user_name='john')
         assert len(r) == 0
 
     @testing.emits_warning('.*empty sequence.*')
-    @testing.fails_on('firebird', 'uses sql-92 bind rules')
     def test_literal_in(self):
         """similar to test_bind_in but use a bind with a value."""
 
             returning=(1, 5)
         )
 
-    @testing.fails_on('mssql', 
+    @testing.fails_on('mssql',
         "lowercase table doesn't support identity insert disable")
     def test_direct_params(self):
         t = self._fixture()
             inserted_primary_key=[]
         )
 
-    @testing.fails_on('mssql', 
+    @testing.fails_on('mssql',
         "lowercase table doesn't support identity insert disable")
     @testing.requires.returning
     def test_direct_params_returning(self):

test/sql/test_types.py

     #                    lambda: testing.db_spec("postgresql")(testing.db),
     #                    "pg8000 and psycopg2 both have issues here in py3k"
     #                    )
-    @testing.skip_if(lambda: testing.db_spec('mssql+mxodbc'), 
+    @testing.skip_if(lambda: testing.db_spec('mssql+mxodbc'),
         "unsupported behavior")
     def test_ignoring_unicode_error(self):
         """checks String(unicode_error='ignore') is passed to underlying codec."""
     __excluded_on__ = (
         ('mysql', '<', (4, 1, 1)),  # screwy varbinary types
     )
-    
+
     @classmethod
     def setup_class(cls):
         global binary_table, MyPickleType, metadata
                     return value / 10
                 return process
             def adapt_operator(self, op):
-                return {operators.add:operators.sub, 
+                return {operators.add:operators.sub,
                     operators.sub:operators.add}.get(op, op)
 
         class MyTypeDec(types.TypeDecorator):
     def teardown_class(cls):
         users_with_date.drop()
 
-    def testdate(self):
+    def test_date_roundtrip(self):
         global insert_data
 
         l = map(tuple,
         self.assert_(l == insert_data,
                      'DateTest mismatch: got:%s expected:%s' % (l, insert_data))
 
-    def testtextdate(self):
+    def test_text_date_roundtrip(self):
         x = testing.db.execute(text(
             "select user_datetime from query_users_with_date",
-            typemap={'user_datetime':DateTime})).fetchall()
+            typemap={'user_datetime': DateTime})).fetchall()
 
         self.assert_(isinstance(x[0][0], datetime.datetime))
 
             bindparams=[bindparam('somedate', type_=types.DateTime)]),
             somedate=datetime.datetime(2005, 11, 10, 11, 52, 35)).fetchall()
 
-    def testdate2(self):
+    def test_date_mixdatetime_roundtrip(self):
         meta = MetaData(testing.db)
         t = Table('testdate', meta,
-                  Column('id', Integer,
+                    Column('id', Integer,
                          Sequence('datetest_id_seq', optional=True),
                          primary_key=True),
-                Column('adate', Date), Column('adatetime', DateTime))
+                    Column('adate', Date),
+                    Column('adatetime', DateTime))
         t.create(checkfirst=True)
         try:
             d1 = datetime.date(2007, 10, 30)
 
             # test mismatched date/datetime
             t.insert().execute(adate=d2, adatetime=d2)
-            eq_(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
-            eq_(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
+            eq_(
+                select([t.c.adate, t.c.adatetime], t.c.adate == d1)\
+                    .execute().fetchall(),
+                [(d1, d2)])
+            eq_(
+                select([t.c.adate, t.c.adatetime], t.c.adate == d1)\
+                    .execute().fetchall(),
+                [(d1, d2)])
 
         finally:
             t.drop(checkfirst=True)
     )
     @testing.fails_on('postgresql+pg8000',
         "pg-8000 does native decimal but truncates the decimals.")
+    @testing.fails_on("firebird",
+        "database and/or driver truncates decimal places."
+        )
     def test_numeric_no_decimal(self):
         numbers = set([
             decimal.Decimal("1.000")
         assert isinstance(val, float)
 
         # some DBAPIs have unusual float handling
-        if testing.against('oracle+cx_oracle', 'mysql+oursql'):
+        if testing.against('oracle+cx_oracle', 'mysql+oursql', 'firebird'):
             eq_(round_decimal(val, 3), 46.583)
         else:
             eq_(val, 46.583)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.