本文整理汇总了Python中psycopg2.extensions.b函数的典型用法代码示例。如果您正苦于以下问题：Python b函数的具体用法？Python b怎么用？Python b使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了b函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_mogrify_unicode
def test_mogrify_unicode(self):
conn = self.conn
cur = conn.cursor()
# test consistency between execute and mogrify.
# unicode query containing only ascii data
cur.execute(u"SELECT 'foo';")
self.assertEqual('foo', cur.fetchone()[0])
self.assertEqual(b("SELECT 'foo';"), cur.mogrify(u"SELECT 'foo';"))
conn.set_client_encoding('UTF8')
snowman = u"\u2603"
# unicode query with non-ascii data
cur.execute(u"SELECT '%s';" % snowman)
self.assertEqual(snowman.encode('utf8'), b(cur.fetchone()[0]))
self.assertEqual(("SELECT '%s';" % snowman).encode('utf8'),
cur.mogrify(u"SELECT '%s';" % snowman).replace(b("E'"), b("'")))
# unicode args
cur.execute("SELECT %s;", (snowman,))
self.assertEqual(snowman.encode("utf-8"), b(cur.fetchone()[0]))
self.assertEqual(("SELECT '%s';" % snowman).encode('utf8'),
cur.mogrify("SELECT %s;", (snowman,)).replace(b("E'"), b("'")))
# unicode query and args
cur.execute(u"SELECT %s;", (snowman,))
self.assertEqual(snowman.encode("utf-8"), b(cur.fetchone()[0]))
self.assertEqual(("SELECT '%s';" % snowman).encode('utf8'),
cur.mogrify(u"SELECT %s;", (snowman,)).replace(b("E'"), b("'")))
开发者ID:Akhilesh05，项目名称:tutize，代码行数:31，代码来源:test_cursor.py
示例2: getquoted
def getquoted(self):
if self.name is None:
raise NotImplementedError(
'RangeAdapter must be subclassed overriding its name '
'or the getquoted() method')
r = self.adapted
if r.isempty:
return b("'empty'::" + self.name)
if r.lower is not None:
a = adapt(r.lower)
if hasattr(a, 'prepare'):
a.prepare(self._conn)
lower = a.getquoted()
else:
lower = b'NULL'
if r.upper is not None:
a = adapt(r.upper)
if hasattr(a, 'prepare'):
a.prepare(self._conn)
upper = a.getquoted()
else:
upper = b'NULL'
return b(self.name + '(') + lower + b', ' + upper \
+ b(", '%s')" % r._bounds)
开发者ID:Abixer，项目名称:croomcroom，代码行数:28，代码来源:_range.py
示例3: test_adapt_bytes
def test_adapt_bytes(self):
snowman = u"\u2603"
self.conn.set_client_encoding('utf8')
a = psycopg2.extensions.QuotedString(snowman.encode('utf8'))
a.prepare(self.conn)
q = a.getquoted()
self.assert_(q in (b("'\xe2\x98\x83'"), b("E'\xe2\x98\x83'")), q)
开发者ID:Aisling-Dempsey，项目名称:pencil_me_in，代码行数:7，代码来源:test_quote.py
示例4: test_read_binary
def test_read_binary(self):
lo = self.conn.lobject()
length = lo.write(b("some data"))
lo.close()
lo = self.conn.lobject(lo.oid, "rb")
x = lo.read(4)
self.assertEqual(type(x), type(b('')))
self.assertEqual(x, b("some"))
self.assertEqual(lo.read(), b(" data"))
开发者ID:44maagnum，项目名称:princetron_website，代码行数:10，代码来源:test_lobject.py
示例5: _getquoted_9
def _getquoted_9(self):
"""Use the hstore(text[], text[]) function."""
if not self.wrapped:
return b("''::hstore")
k = _ext.adapt(self.wrapped.keys())
k.prepare(self.conn)
v = _ext.adapt(self.wrapped.values())
v.prepare(self.conn)
return b("hstore(") + k.getquoted() + b(", ") + v.getquoted() + b(")")
开发者ID:Aisling-Dempsey，项目名称:pencil_me_in，代码行数:10，代码来源:extras.py
示例6: adapt_range
def adapt_range(pgrange, pyrange):
if not isinstance(pyrange, range_):
raise ValueError((
"Trying to adapt range {range.__class__.__name__} which does not "
"extend base range type.").format(range=pyrange))
if not pyrange:
return AsIs("'empty'::" + pgrange)
lower = b("NULL")
if not pyrange.lower_inf:
lower = adapt(pyrange.lower).getquoted()
upper = b("NULL")
if not pyrange.upper_inf:
upper = adapt(pyrange.upper).getquoted()
return AsIs(b"".join([
b(pgrange),
b("("),
lower,
b(", "),
upper,
b(", '"),
b("[" if pyrange.lower_inc else "("),
b("]" if pyrange.upper_inc else ")"),
b("')")
]).decode("utf8"))
开发者ID:pombredanne，项目名称:psycospans，代码行数:28，代码来源:_utils.py
示例7: test_set_encoding
def test_set_encoding(self):
# Note: this works-ish mostly in case when the standard db connection
# we test with is utf8, otherwise the encoding chosen by PQescapeString
# may give bad results.
from psycopg2.extensions import adapt
snowman = u"\u2603"
a = adapt(snowman)
a.encoding = 'utf8'
self.assertEqual(a.encoding, 'utf8')
q = a.getquoted()
self.assert_(q in (b("'\xe2\x98\x83'"), b("E'\xe2\x98\x83'")), q)
开发者ID:Aisling-Dempsey，项目名称:pencil_me_in，代码行数:11，代码来源:test_quote.py
示例8: test_connection_wins_anyway
def test_connection_wins_anyway(self):
from psycopg2.extensions import adapt
snowman = u"\u2603"
a = adapt(snowman)
a.encoding = 'latin9'
self.conn.set_client_encoding('utf8')
a.prepare(self.conn)
self.assertEqual(a.encoding, 'utf_8')
q = a.getquoted()
self.assert_(q in (b("'\xe2\x98\x83'"), b("E'\xe2\x98\x83'")), q)
开发者ID:Aisling-Dempsey，项目名称:pencil_me_in，代码行数:12，代码来源:test_quote.py
示例9: test_export
def test_export(self):
lo = self.conn.lobject()
lo.write(b("some data"))
self.tmpdir = tempfile.mkdtemp()
filename = os.path.join(self.tmpdir, "data.txt")
lo.export(filename)
self.assertTrue(os.path.exists(filename))
f = open(filename, "rb")
try:
self.assertEqual(f.read(), b("some data"))
finally:
f.close()
开发者ID:44maagnum，项目名称:princetron_website，代码行数:13，代码来源:test_lobject.py
示例10: test_inet_conform
def test_inet_conform(self):
from psycopg2.extras import Inet
i = Inet("192.168.1.0/24")
a = psycopg2.extensions.adapt(i)
a.prepare(self.conn)
self.assertEqual(filter_scs(self.conn, b("E'192.168.1.0/24'::inet")), a.getquoted())
# adapts ok with unicode too
i = Inet(u"192.168.1.0/24")
a = psycopg2.extensions.adapt(i)
a.prepare(self.conn)
self.assertEqual(filter_scs(self.conn, b("E'192.168.1.0/24'::inet")), a.getquoted())
开发者ID:hayalifabrika，项目名称:living-galapagos，代码行数:13，代码来源:test_types_extras.py
示例11: test_none_in_record
def test_none_in_record(self):
curs = self.conn.cursor()
s = curs.mogrify("SELECT %s;", [(42, None)])
self.assertEqual(b("SELECT (42, NULL);"), s)
curs.execute("SELECT %s;", [(42, None)])
d = curs.fetchone()[0]
self.assertEqual("(42,)", d)
开发者ID:hayalifabrika，项目名称:living-galapagos，代码行数:7，代码来源:test_types_extras.py
示例12: test_full_escaped_octal
def test_full_escaped_octal(self):
buf = ''.join(("\\%03o" % i) for i in range(256))
rv = self.cast(b(buf))
if sys.version_info[0] < 3:
self.assertEqual(rv, ''.join(map(chr, list(range(256)))))
else:
self.assertEqual(rv, bytes(list(range(256))))
开发者ID:ninoch，项目名称:Notes_Manager，代码行数:7，代码来源:test_types_basic.py
示例13: test_adapt_8
def test_adapt_8(self):
if self.conn.server_version >= 90000:
return self.skipTest("skipping dict adaptation with PG pre-9 syntax")
from psycopg2.extras import HstoreAdapter
o = {"a": "1", "b": "'", "c": None}
if self.conn.encoding == "UTF8":
o["d"] = u"\xe0"
a = HstoreAdapter(o)
a.prepare(self.conn)
q = a.getquoted()
self.assert_(q.startswith(b("((")), q)
ii = q[1:-1].split(b("||"))
ii.sort()
self.assertEqual(len(ii), len(o))
self.assertEqual(ii[0], filter_scs(self.conn, b("(E'a' => E'1')")))
self.assertEqual(ii[1], filter_scs(self.conn, b("(E'b' => E'''')")))
self.assertEqual(ii[2], filter_scs(self.conn, b("(E'c' => NULL)")))
if "d" in o:
encc = u"\xe0".encode(psycopg2.extensions.encodings[self.conn.encoding])
self.assertEqual(ii[3], filter_scs(self.conn, b("(E'd' => E'") + encc + b("')")))
开发者ID:hayalifabrika，项目名称:living-galapagos，代码行数:25，代码来源:test_types_extras.py
示例14: test_full_hex
def test_full_hex(self, upper=False):
buf = ''.join(("%02x" % i) for i in range(256))
if upper: buf = buf.upper()
buf = '\\x' + buf
rv = self.cast(b(buf))
if sys.version_info[0] < 3:
self.assertEqual(rv, ''.join(map(chr, list(range(256)))))
else:
self.assertEqual(rv, bytes(list(range(256))))
开发者ID:ninoch，项目名称:Notes_Manager，代码行数:9，代码来源:test_types_basic.py
示例15: test_import
def test_import(self):
self.tmpdir = tempfile.mkdtemp()
filename = os.path.join(self.tmpdir, "data.txt")
fp = open(filename, "wb")
fp.write(b("some data"))
fp.close()
lo = self.conn.lobject(0, "r", 0, filename)
self.assertEqual(lo.read(), "some data")
开发者ID:44maagnum，项目名称:princetron_website，代码行数:9，代码来源:test_lobject.py
示例16: test_mogrify_decimal_explodes
def test_mogrify_decimal_explodes(self):
# issue #7: explodes on windows with python 2.5 and psycopg 2.2.2
try:
from decimal import Decimal
except:
return
conn = self.conn
cur = conn.cursor()
self.assertEqual(b("SELECT 10.3;"), cur.mogrify("SELECT %s;", (Decimal("10.3"),)))
开发者ID:BobbyJoeSmith3，项目名称:swampr，代码行数:10，代码来源:test_cursor.py
示例17: test_read
def test_read(self):
lo = self.conn.lobject()
length = lo.write(b("some data"))
lo.close()
lo = self.conn.lobject(lo.oid)
x = lo.read(4)
self.assertEqual(type(x), type(''))
self.assertEqual(x, "some")
self.assertEqual(lo.read(), " data")
开发者ID:44maagnum，项目名称:princetron_website，代码行数:10，代码来源:test_lobject.py
示例18: test_adapt_subtype_3
def test_adapt_subtype_3(self):
from psycopg2.extensions import adapt, register_adapter, AsIs
class A: pass
class B(A): pass
register_adapter(A, lambda a: AsIs("a"))
try:
self.assertEqual(b("a"), adapt(B()).getquoted())
finally:
del psycopg2.extensions.adapters[A, psycopg2.extensions.ISQLQuote]
开发者ID:ARRDM，项目名称:djando，代码行数:11，代码来源:test_types_basic.py
示例19: getquoted
def getquoted(self):
if self.name is None:
raise NotImplementedError(
"RangeAdapter must be subclassed overriding its name " "or the getquoted() method"
)
r = self.adapted
if r.isempty:
return b("'empty'::" + self.name)
if r.lower is not None:
a = adapt(r.lower)
if hasattr(a, "prepare"):
a.prepare(self._conn)
lower = a.getquoted()
else:
lower = b("NULL")
if r.upper is not None:
a = adapt(r.upper)
if hasattr(a, "prepare"):
a.prepare(self._conn)
upper = a.getquoted()
else:
upper = b("NULL")
return b(self.name + "(") + lower + b(", ") + upper + b(", '%s')" % r._bounds)
开发者ID:MNI-NIL，项目名称:NIL-MNI.github.io，代码行数:27，代码来源:_range.py
示例20: test_binary
def test_binary(self):
data = b("""some data with \000\013 binary
stuff into, 'quotes' and \\ a backslash too.
""")
if sys.version_info[0] < 3:
data += "".join(map(chr, range(256)))
else:
data += bytes(range(256))
curs = self.conn.cursor()
curs.execute("SELECT %s::bytea;", (psycopg2.Binary(data),))
if sys.version_info[0] < 3:
res = str(curs.fetchone()[0])
else:
res = curs.fetchone()[0].tobytes()
if res[0] in (b('x'), ord(b('x'))) and self.conn.server_version >= 90000:
return self.skipTest(
"bytea broken with server >= 9.0, libpq < 9")
self.assertEqual(res, data)
self.assert_(not self.conn.notices)
开发者ID:13lcp2000，项目名称:recetario4-4，代码行数:22，代码来源:test_quote.py
注：本文中的psycopg2.extensions.b函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论