Changeset 662 for trunk


Ignore:
Timestamp:
Dec 10, 2015, 6:36:49 AM (4 years ago)
Author:
cito
Message:

insert() and update() should convert bytes to bytea

As suggested by D'Arcy on the mailing list.

Location:
trunk/module
Files:
2 edited

Legend:

Unmodified
Added
Removed
  • trunk/module/pg.py

    r656 r662  
    384384        return d
    385385
     386    if bytes is str:  # Python < 3.0
     387        """Quote bytes value."""
     388
     389        def _quote_bytea(self, d):
     390            return "'%s'" % self.escape_bytea(d)
     391
     392    else:
     393
     394        def _quote_bytea(self, d):
     395            return "'%s'" % self.escape_bytea(d).decode('ascii')
     396
    386397    _quote_funcs = dict(  # quote methods for each type
    387398        text=_quote_text, bool=_quote_bool, date=_quote_date,
    388399        int=_quote_num, num=_quote_num, float=_quote_num,
    389         money=_quote_money)
     400        money=_quote_money, bytea=_quote_bytea)
    390401
    391402    def _quote(self, d, t):
     
    689700                elif typ.startswith('money'):
    690701                    typ = 'money'
     702                elif typ.startswith('bytea'):
     703                    typ = 'bytea'
    691704                else:
    692705                    typ = 'text'
  • trunk/module/tests/test_classic_dbwrapper.py

    r648 r662  
    10721072        query = self.db.query
    10731073        query('drop table if exists bytea_test')
    1074         query('create table bytea_test ('
    1075             'data bytea)')
     1074        query('create table bytea_test (n smallint primary key, data bytea)')
    10761075        s = b"It's all \\ kinds \x00 of\r nasty \xff stuff!\n"
    10771076        r = self.db.escape_bytea(s)
    1078         query('insert into bytea_test values($1)', (r,))
    1079         r = query('select * from bytea_test').getresult()
    1080         self.assertTrue(len(r) == 1)
     1077        query('insert into bytea_test values(3,$1)', (r,))
     1078        r = query('select * from bytea_test where n=3').getresult()
     1079        self.assertEqual(len(r), 1)
    10811080        r = r[0]
    1082         self.assertTrue(len(r) == 1)
     1081        self.assertEqual(len(r), 2)
     1082        self.assertEqual(r[0], 3)
     1083        r = r[1]
     1084        self.assertIsInstance(r, str)
     1085        r = self.db.unescape_bytea(r)
     1086        self.assertIsInstance(r, bytes)
     1087        self.assertEqual(r, s)
     1088        query('drop table bytea_test')
     1089
     1090    def testInsertUpdateBytea(self):
     1091        query = self.db.query
     1092        query('drop table if exists bytea_test')
     1093        query('create table bytea_test (n smallint primary key, data bytea)')
     1094        # insert as bytes
     1095        s = b"It's all \\ kinds \x00 of\r nasty \xff stuff!\n"
     1096        r = self.db.insert('bytea_test', n=5, data=s)
     1097        self.assertIsInstance(r, dict)
     1098        self.assertIn('n', r)
     1099        self.assertEqual(r['n'], 5)
     1100        self.assertIn('data', r)
     1101        r = r['data']
     1102        # the following two lines should be removed once insert()
     1103        # will be enhanced to adapt the types of return values
     1104        self.assertIsInstance(r, str)
     1105        r = self.db.unescape_bytea(r)
     1106        self.assertIsInstance(r, bytes)
     1107        self.assertEqual(r, s)
     1108        # update as bytes
     1109        s += b"and now even more \x00 nasty \t stuff!\f"
     1110        r = self.db.update('bytea_test', n=5, data=s)
     1111        self.assertIsInstance(r, dict)
     1112        self.assertIn('n', r)
     1113        self.assertEqual(r['n'], 5)
     1114        self.assertIn('data', r)
     1115        r = r['data']
     1116        # the following two lines should be removed once update()
     1117        # will be enhanced to adapt the types of return values
     1118        self.assertIsInstance(r, str)
     1119        r = self.db.unescape_bytea(r)
     1120        self.assertIsInstance(r, bytes)
     1121        self.assertEqual(r, s)
     1122        r = query('select * from bytea_test where n=5').getresult()
     1123        self.assertEqual(len(r), 1)
    10831124        r = r[0]
     1125        self.assertEqual(len(r), 2)
     1126        self.assertEqual(r[0], 5)
     1127        r = r[1]
     1128        self.assertIsInstance(r, str)
    10841129        r = self.db.unescape_bytea(r)
    10851130        self.assertIsInstance(r, bytes)
Note: See TracChangeset for help on using the changeset viewer.