from ..orm.attributes import QueryableAttribute
from .. import Table, Column
from ..engine import Engine
-from ..util import pickle
+from ..util import pickle, text_type
import re
import base64
# Py3K
id = "mapperprop:" + b64encode(pickle.dumps(obj.parent.class_)) + \
":" + obj.key
elif isinstance(obj, Table):
- id = "table:" + str(obj)
+ id = "table:" + text_type(obj.key)
elif isinstance(obj, Column) and isinstance(obj.table, Table):
- id = "column:" + str(obj.table) + ":" + obj.key
+ id = "column:" + text_type(obj.table.key) + ":" + text_type(obj.key)
elif isinstance(obj, Session):
id = "session:"
elif isinstance(obj, Engine):
return None
def persistent_load(id):
- m = our_ids.match(str(id))
+ m = our_ids.match(text_type(id))
if not m:
return None
else:
c = clause.compile(dialect=dialect, **kw)
param_str = repr(getattr(c, 'params', {}))
- # Py3K
- #param_str = param_str.encode('utf-8').decode('ascii', 'ignore')
- print "\nSQL String:\n" + str(c) + param_str
+ if util.py3k:
+ param_str = param_str.encode('utf-8').decode('ascii', 'ignore')
- cc = re.sub(r'[\n\t]', '', str(c))
+ print("\nSQL String:\n" + util.text_type(c).encode('utf-8') + param_str)
+
+ cc = re.sub(r'[\n\t]', '', util.text_type(c))
eq_(cc, result, "%r != %r on dialect %r" % (cc, result, dialect))
+# coding: utf-8
from sqlalchemy.ext import serializer
from sqlalchemy import testing
from sqlalchemy import Integer, String, ForeignKey, select, \
- desc, func, util
+ desc, func, util, MetaData
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.schema import Column
from sqlalchemy.orm import relationship, sessionmaker, scoped_session, \
class_mapper, mapper, joinedload, configure_mappers, aliased
-from sqlalchemy.testing import eq_
+from sqlalchemy.testing import eq_, AssertsCompiledSQL
+from sqlalchemy.util import u, ue
+from sqlalchemy.engine import default
from sqlalchemy.testing import fixtures
users = addresses = Session = None
-class SerializeTest(fixtures.MappedTest):
+class SerializeTest(AssertsCompiledSQL, fixtures.MappedTest):
run_setup_mappers = 'once'
run_inserts = 'once'
x = serializer.loads(ser, users.metadata)
eq_(str(r), str(x))
+ def test_unicode(self):
+ m = MetaData()
+ t = Table(ue('\u6e2c\u8a66'), m,
+ Column(ue('\u6e2c\u8a66_id'), Integer))
+
+ expr = select([t]).where(t.c[ue('\u6e2c\u8a66_id')] == 5)
+
+ expr2 = serializer.loads(serializer.dumps(expr, -1), m)
+
+ self.assert_compile(
+ expr2,
+ ue('SELECT "\u6e2c\u8a66"."\u6e2c\u8a66_id" FROM "\u6e2c\u8a66" '
+ 'WHERE "\u6e2c\u8a66"."\u6e2c\u8a66_id" = :\u6e2c\u8a66_id_1'),
+ dialect=default.DefaultDialect()
+ )
+
if __name__ == '__main__':
testing.main()