Commits

Mike Bayer committed 31aa8d5

added convert_unicode flag to engine + unittest, does unicode in/out translation on all string/char values when set

  • Participants
  • Parent commits 86bfd60

Comments (0)

Files changed (5)

doc/build/content/dbengine.myt

         <li>default_ordering=False : if True, table objects and associated joins and aliases will generate information used for ordering by primary keys (or OIDs, if the database supports OIDs).  This information is used by the Mapper system to when it constructs select queries to supply a default ordering to mapped objects.</li>
         <li>use_ansi=True : used only by Oracle;  when False, the Oracle driver attempts to support a particular "quirk" of some Oracle databases, that the LEFT OUTER JOIN SQL syntax is not supported, and the "Oracle join" syntax of using <% "<column1>(+)=<column2>" |h%> must be used in order to achieve a LEFT OUTER JOIN.  Its advised that the Oracle database be configured to have full ANSI support instead of using this feature.</li>
         <li>use_oids=False : used only by Postgres, will enable the column name "oid" as the object ID column.  Postgres as of 8.1 has object IDs disabled by default.</li>
+        <li>convert_unicode=False : if set to True, all String/character based types will convert Unicode values to raw byte values going into the database, and all raw byte values to Python Unicode coming out in result sets.  This is an engine-wide method to provide unicode across the board.  For unicode conversion on a column-by-column level, use the Unicode column type instead.</li>
     </ul>
     </&>
     <&|doclib.myt:item, name="proxy", description="Using the Proxy Engine" &>

lib/sqlalchemy/databases/postgres.py

         return value
     def get_col_spec(self):
         return "TIME"
+
 class PGText(sqltypes.TEXT):
     def get_col_spec(self):
         return "TEXT"

lib/sqlalchemy/engine.py

     SQLEngines are constructed via the create_engine() function inside this package.
     """
     
-    def __init__(self, pool=None, echo=False, logger=None, default_ordering=False, echo_pool=False, echo_uow=False, **params):
+    def __init__(self, pool=None, echo=False, logger=None, default_ordering=False, echo_pool=False, echo_uow=False, convert_unicode=False, **params):
         """constructs a new SQLEngine.   SQLEngines should be constructed via the create_engine()
         function which will construct the appropriate subclass of SQLEngine."""
         # get a handle on the connection pool via the connect arguments
         self.default_ordering=default_ordering
         self.echo = echo
         self.echo_uow = echo_uow
+        self.convert_unicode = convert_unicode
         self.context = util.ThreadLocal(raiseerror=False)
         self.tables = {}
         self._ischema = None

lib/sqlalchemy/types.py

             return TEXT()
         else:
             return self
+    def convert_bind_param(self, value, engine):
+        if not engine.convert_unicode or not isinstance(value, unicode):
+            return value
+        else:
+            return value.encode('utf-8')
+    def convert_result_value(self, value, engine):
+        if not engine.convert_unicode or isinstance(value, unicode):
+            return value
+        else:
+            return value.decode('utf-8')
 
 class Unicode(TypeDecorator,String):
     def __init__(self, length=None):

test/testtypes.py

             self.echo("its sqlite !")
         else:
             self.assert_(not isinstance(x['plain_data'], unicode) and x['plain_data'] == rawdata)
-            
+    def testengineparam(self):
+        """tests engine-wide unicode conversion"""
+        prev_unicode = db.engine.convert_unicode
+        try:
+            db.engine.convert_unicode = True
+            rawdata = 'Alors vous imaginez ma surprise, au lever du jour, quand une dr\xc3\xb4le de petit voix m\xe2\x80\x99a r\xc3\xa9veill\xc3\xa9. Elle disait: \xc2\xab S\xe2\x80\x99il vous pla\xc3\xaet\xe2\x80\xa6 dessine-moi un mouton! \xc2\xbb\n'
+            unicodedata = rawdata.decode('utf-8')
+            unicode_table.insert().execute(unicode_data=unicodedata, plain_data=rawdata)
+            x = unicode_table.select().execute().fetchone()
+            self.echo(repr(x['unicode_data']))
+            self.echo(repr(x['plain_data']))
+            self.assert_(isinstance(x['unicode_data'], unicode) and x['unicode_data'] == unicodedata)
+            self.assert_(isinstance(x['plain_data'], unicode) and x['plain_data'] == unicodedata)
+        finally:
+            db.engine.convert_unicode = prev_unicode
     
 class BinaryTest(AssertMixin):
     def setUpAll(self):