我们从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_varchar(self): """ A varchar type raises an error when too many characters are passed. """ NoPk = self.db['no_pk'] # bar is short enough NoPk(foo='abcdefghij').flush() self.assertRaises(psycopg2.DataError, NoPk(foo='abcdefghijk').flush)
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 testArrayMalformed(self): curs = self.conn.cursor() ss = ['', '{', '{}}', '{' * 20 + '}' * 20] for s in ss: self.assertRaises(psycopg2.DataError, psycopg2.extensions.STRINGARRAY, s.encode('utf8'), curs)
def test_parse_incomplete_date(self): self.assertRaises(psycopg2.DataError, self.DATE, '2007', self.curs) self.assertRaises(psycopg2.DataError, self.DATE, '2007-01', self.curs)
def test_parse_incomplete_time(self): self.assertRaises(psycopg2.DataError, self.TIME, '13', self.curs) self.assertRaises(psycopg2.DataError, self.TIME, '13:30', self.curs)
def test_copy_rowcount_error(self): curs = self.conn.cursor() curs.execute("insert into tcopy (data) values ('fff')") self.assertEqual(curs.rowcount, 1) self.assertRaises(psycopg2.DataError, curs.copy_from, StringIO('aaa\nbbb\nccc\n'), 'tcopy') self.assertEqual(curs.rowcount, -1)
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', 23), ('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", (float("inf"),)) self.failUnless(str(s) == "inf", "wrong float quoting: " + str(s)) self.failUnless(type(s) == float, "wrong float conversion: " + repr(s)) s = self.execute("SELECT %s AS foo", (float("-inf"),)) self.failUnless(str(s) == "-inf", "wrong float quoting: " + str(s))
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.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 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 testArrayMalformed(self): curs = self.conn.cursor() ss = ['', '{', '{}}', '{' * 20 + '}' * 20] for s in ss: self.assertRaises(psycopg2.DataError, psycopg2.extensions.STRINGARRAY, b(s), curs)
def test_parse_incomplete_datetime(self): self.assertRaises(psycopg2.DataError, self.DATETIME, '2007', self.curs) self.assertRaises(psycopg2.DataError, self.DATETIME, '2007-01', self.curs) self.assertRaises(psycopg2.DataError, self.DATETIME, '2007-01-01 13', self.curs) self.assertRaises(psycopg2.DataError, self.DATETIME, '2007-01-01 13:30', self.curs)
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)
