Python psycopg2 模块,DataError() 实例源码
我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用psycopg2.DataError()。
def test_start_and_recover_from_error(self):
conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
if conn is None:
return
cur = conn.cursor()
self.create_replication_slot(cur, output_plugin='test_decoding')
# try with invalid options
cur.start_replication(
slot_name=self.slot, options={'invalid_param': 'value'})
def consume(msg):
pass
# we don't see the error from the server before we try to read the data
self.assertRaises(psycopg2.DataError, cur.consume_stream, consume)
# try with correct command
cur.start_replication(slot_name=self.slot)
def test_start_and_recover_from_error(self):
conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
if conn is None:
return
cur = conn.cursor()
self.create_replication_slot(cur, consume)
# try with correct command
cur.start_replication(slot_name=self.slot)
def test_start_and_recover_from_error(self):
conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
if conn is None:
return
cur = conn.cursor()
self.create_replication_slot(cur, consume)
# try with correct command
cur.start_replication(slot_name=self.slot)
def test_start_and_recover_from_error(self):
conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
if conn is None:
return
cur = conn.cursor()
self.create_replication_slot(cur, consume)
# try with correct command
cur.start_replication(slot_name=self.slot)
def test_start_and_recover_from_error(self):
conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
if conn is None:
return
cur = conn.cursor()
self.create_replication_slot(cur, consume)
# try with correct command
cur.start_replication(slot_name=self.slot)
def testFloatInf(self):
try:
self.execute("select 'inf'::float")
except psycopg2.DataError:
return self.skipTest("inf::float not available on the server")
except ValueError:
return self.skipTest("inf not available on this platform")
s = self.execute("SELECT %s AS foo", (float("inf"),))
self.assertTrue(str(s) == "inf", "wrong float quoting: " + str(s))
self.assertTrue(type(s) == float, "wrong float conversion: " + repr(s))
s = self.execute("SELECT %s AS foo", (float("-inf"),))
self.assertTrue(str(s) == "-inf", "wrong float quoting: " + str(s))
def test_parse_incomplete_time(self):
self.assertRaises(psycopg2.DataError, self.TIME, '13', '13:30', self.curs)
def test_with_error_db(self):
def f():
with self.conn as conn:
curs = conn.cursor()
curs.execute("insert into test_with values ('a')")
self.assertRaises(psycopg2.DataError, f)
self.assertEqual(self.conn.status, ext.STATUS_READY)
self.assertTrue(not self.conn.closed)
curs = self.conn.cursor()
curs.execute("select * from test_with")
self.assertEqual(curs.fetchall(), [])
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
try:
with self.conn as conn:
with conn.cursor('named') as cur:
cur.execute("select 1/0")
cur.fetchone()
except psycopg2.DataError as e:
self.assertEqual(e.pgcode, '22012')
else:
self.fail("where is my exception?")
def test_wrong_schema(self):
oid = self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
from psycopg2.extras import CompositeCaster
c = CompositeCaster('type_ii', oid, [('a', 23), ('b', ('c', 23)])
curs = self.conn.cursor()
psycopg2.extensions.register_type(c.typecaster, curs)
curs.execute("select (1,2)::type_ii")
self.assertRaises(psycopg2.DataError, curs.fetchone)
def testFloatInf(self):
try:
self.execute("select 'inf'::float")
except psycopg2.DataError:
return self.skipTest("inf::float not available on the server")
except ValueError:
return self.skipTest("inf not available on this platform")
s = self.execute("SELECT %s AS foo",))
self.failUnless(str(s) == "inf", "wrong float quoting: " + str(s))
self.failUnless(type(s) == float,))
self.failUnless(str(s) == "-inf", "wrong float quoting: " + str(s))
def testArrayMalformed(self):
curs = self.conn.cursor()
ss = ['', curs)
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_parse_incomplete_time(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_with_error_db(self):
def f():
with self.conn as conn:
curs = conn.cursor()
curs.execute("insert into test_with values ('a')")
self.assertRaises(psycopg2.DataError, ext.STATUS_READY)
self.assert_(not self.conn.closed)
curs = self.conn.cursor()
curs.execute("select * from test_with")
self.assertEqual(curs.fetchall(), [])
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
try:
with self.conn as conn:
with conn.cursor('named') as cur:
cur.execute("select 1/0")
cur.fetchone()
except psycopg2.DataError, e:
self.assertEqual(e.pgcode, '22012')
else:
self.fail("where is my exception?")
def test_wrong_schema(self):
oid = self._create_type("type_ii", curs.fetchone)
def handle_listing(listing_object, results_item):
try:
new_listing_object = extract_listing_data(results_item)
updated_listing_object = update_core_listing_data(listing_object,
new_listing_object)
updated_listing_object.save()
except (ValueError, DataError, PG_DataError):
logger.debug(traceback.print_exc(limit=None))
raise ListingException()
# Handle premature exits
def testFloatInf(self):
try:
self.execute("select 'inf'::float")
except psycopg2.DataError:
return self.skipTest("inf::float not available on the server")
except ValueError:
return self.skipTest("inf not available on this platform")
s = self.execute("SELECT %s AS foo", "wrong float quoting: " + str(s))
def testArrayMalformed(self):
curs = self.conn.cursor()
ss = ['', curs)
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_parse_incomplete_time(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_with_error_db(self):
def f():
with self.conn as conn:
curs = conn.cursor()
curs.execute("insert into test_with values ('a')")
self.assertRaises(psycopg2.DataError, [])
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
try:
with self.conn as conn:
with conn.cursor('named') as cur:
cur.execute("select 1/0")
cur.fetchone()
except psycopg2.DataError as e:
self.assertEqual(e.pgcode, '22012')
else:
self.fail("where is my exception?")
def test_wrong_schema(self):
oid = self._create_type("type_ii", curs.fetchone)
def testFloatInf(self):
try:
self.execute("select 'inf'::float")
except psycopg2.DataError:
return self.skipTest("inf::float not available on the server")
except ValueError:
return self.skipTest("inf not available on this platform")
s = self.execute("SELECT %s AS foo", "wrong float quoting: " + str(s))
self.failUnless(type(s) == float, "wrong float quoting: " + str(s))
def testArrayMalformed(self):
curs = self.conn.cursor()
ss = ['', b(s), curs)
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_parse_incomplete_time(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_parse_incomplete_datetime(self):
self.assertRaises(psycopg2.DataError,
self.DATETIME, '2007-01-01 13', '2007-01-01 13:30', self.curs)
def test_with_error_db(self):
def f():
with self.conn as conn:
curs = conn.cursor()
curs.execute("insert into test_with values ('a')")
self.assertRaises(psycopg2.DataError, [])
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
try:
with self.conn as conn:
with conn.cursor('named') as cur:
cur.execute("select 1/0")
cur.fetchone()
except psycopg2.DataError, '22012')
else:
self.fail("where is my exception?")
def test_wrong_schema(self):
oid = self._create_type("type_ii", curs.fetchone)
def parse(self, s, curs):
if s is None:
return None
tokens = self.tokenize(s)
if len(tokens) != len(self.atttypes):
raise psycopg2.DataError(
"expecting %d components for the type %s,%d found instead" %
(len(self.atttypes), self.name, len(tokens)))
values = [ curs.cast(oid, token)
for oid, token in zip(self.atttypes, tokens) ]
return self.make(values)
def testArrayMalformed(self):
curs = self.conn.cursor()
ss = ['', curs)
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_parse_incomplete_time(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_parse_incomplete_datetime(self):
self.assertRaises(psycopg2.DataError, self.curs)
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
try:
with self.conn as conn:
with conn.cursor('named') as cur:
cur.execute("select 1/0")
cur.fetchone()
except psycopg2.DataError, '22012')
else:
self.fail("where is my exception?")
def test_wrong_schema(self):
oid = self._create_type("type_ii", curs.fetchone)
def testFloatInf(self):
try:
self.execute("select 'inf'::float")
except psycopg2.DataError:
return self.skipTest("inf::float not available on the server")
except ValueError:
return self.skipTest("inf not available on this platform")
s = self.execute("SELECT %s AS foo", "wrong float quoting: " + str(s))
def test_parse_incomplete_date(self):
self.assertRaises(psycopg2.DataError, self.curs)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。