def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
python类adapters()的实例源码
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)),
b("""'{"a": 123}'"""))
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_register_range_adapter(self):
from psycopg2.extras import Range, register_range
cur = self.conn.cursor()
cur.execute("create type textrange as range (subtype=text)")
rc = register_range('textrange', 'TextRange', cur)
TextRange = rc.range
self.assert_(issubclass(TextRange, Range))
self.assertEqual(TextRange.__name__, 'TextRange')
r = TextRange('a', 'b', '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
cur.execute("select %s", ([r,r,r],))
rs = cur.fetchone()[0]
self.assertEqual(len(rs), 3)
for r1 in rs:
self.assertEqual(r1.lower, 'a')
self.assertEqual(r1.upper, 'b')
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
# clear the adapters to allow precise count by scripts/refcounter.py
del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self):
cur = self.conn.cursor()
cur.execute("create schema rs")
cur.execute("create type r1 as range (subtype=text)")
cur.execute("create type r2 as range (subtype=text)")
cur.execute("create type rs.r2 as range (subtype=text)")
cur.execute("create type rs.r3 as range (subtype=text)")
cur.execute("savepoint x")
from psycopg2.extras import register_range
ra1 = register_range('r1', 'r1', cur)
ra2 = register_range('r2', 'r2', cur)
rars2 = register_range('rs.r2', 'r2', cur)
rars3 = register_range('rs.r3', 'r3', cur)
self.assertNotEqual(
ra2.typecaster.values[0],
rars2.typecaster.values[0])
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'r3', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
self.assertRaises(psycopg2.ProgrammingError,
register_range, 'rs.r1', 'FailRange', cur)
cur.execute("rollback to savepoint x;")
# clear the adapters to allow precise count by scripts/refcounter.py
for r in [ra1, ra2, rars2, rars3]:
del ext.adapters[r.range, ext.ISQLQuote]
def test_register_on_dict(self):
from psycopg2.extras import Json
psycopg2.extensions.register_adapter(dict, Json)
try:
curs = self.conn.cursor()
obj = {'a': 123}
self.assertEqual(curs.mogrify("%s", (obj,)), b"""'{"a": 123}'""")
finally:
del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
