我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlite3.ProgrammingError()。
def insert(address, port, location='default', protocol='default'): uid = str(uuid.uuid5(uuid.NAMESPACE_DNS,address)) conn = sqlite3.connect(db) try: conn.execute("INSERT INTO PROXY VALUES (?, ?, ?, ?, ?)",[uid,address,port,location,protocol]) print('insert: '+str(address)) pass except sqlite3.IntegrityError as e: conn.execute("UPDATE PROXY SET ADDRESS = ?, PORT = ?, LOCATION = ?, PROTOCOL = ? WHERE UID=?",[address,port,location,protocol,uid]) print('update: '+str(address)) pass except sqlite3.ProgrammingError as e: #TODO pass finally: conn.commit() conn.close()
def log(self, event): """ Log an event into the database. Automatically commits executions. The base log method is fit for the base event class. :param event: event object to commit to db """ try: with self.conn: self.cursor.execute(''' INSERT INTO logs(inputDT, level, log) VALUES(?, ?, ?)''', (event.dt, event.level_str, event.info)) except sqlite3.ProgrammingError: raise ValueError("""diary does not support logging unicode strings into a database in ython2. To avoid this: 1. Ensure your strings are converted to the 'str' type before logging 2. Write your own implementation of DiaryDB that can handle unicode """)
def CheckCursorConstructorCallCheck(self): """ Verifies that cursor methods check whether base class __init__ was called. """ class Cursor(sqlite.Cursor): def __init__(self, con): pass con = sqlite.connect(":memory:") cur = Cursor(con) try: cur.execute("select 4+5").fetchall() self.fail("should have raised ProgrammingError") except sqlite.ProgrammingError: pass except: self.fail("should have raised ProgrammingError")
def CheckConnectionConstructorCallCheck(self): """ Verifies that connection methods check whether base class __init__ was called. """ class Connection(sqlite.Connection): def __init__(self, name): pass con = Connection(":memory:") try: cur = con.cursor() self.fail("should have raised ProgrammingError") except sqlite.ProgrammingError: pass except: self.fail("should have raised ProgrammingError")
def CheckConCursor(self): def run(con, errors): try: cur = con.cursor() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors))
def CheckConCommit(self): def run(con, errors): try: con.commit() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors))
def CheckConClose(self): def run(con, errors): try: con.close() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors))
def CheckCurImplicitBegin(self): def run(cur, errors): try: cur.execute("insert into test(name) values ('a')") errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors))
def CheckCurClose(self): def run(cur, errors): try: cur.close() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors))
def CheckCurExecute(self): def run(cur, errors): try: cur.execute("select name from test") errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] self.cur.execute("insert into test(name) values ('a')") t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors))
def CheckCurIterNext(self): def run(cur, errors): try: row = cur.fetchone() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] self.cur.execute("insert into test(name) values ('a')") self.cur.execute("select name from test") t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors))
def CheckClosedCreateAggregate(self): con = sqlite.connect(":memory:") con.close() class Agg: def __init__(self): pass def step(self, x): pass def finalize(self): return 17 try: con.create_aggregate("foo", 1, Agg) self.fail("Should have raised a ProgrammingError") except sqlite.ProgrammingError: pass except: self.fail("Should have raised a ProgrammingError")
def CheckClosed(self): con = sqlite.connect(":memory:") cur = con.cursor() cur.close() for method_name in ("execute", "executemany", "executescript", "fetchall", "fetchmany", "fetchone"): if method_name in ("execute", "executescript"): params = ("select 4 union select 5",) elif method_name == "executemany": params = ("insert into foo(bar) values (?)", [(3,), (4,)]) else: params = [] try: method = getattr(cur, method_name) method(*params) self.fail("Should have raised a ProgrammingError: method " + method_name) except sqlite.ProgrammingError: pass except: self.fail("Should have raised a ProgrammingError: " + method_name)
def process_item(self, item, spider): try: cursor = self.conn.cursor() cursor.execute("""CREATE TABLE IF NOT EXISTS broad (Title varchar(300) NOT NULL, Url varchar(300) NOT NULL, Date varchar(50) NOT NULL, Content TEXT NOT NULL, Image_Url TEXT NOT NULL)""") record = (item['title'], item['url'], \ item['date'], item['content'], item['image_urls']) cursor.execute('INSERT INTO broad VALUES (?,?,?,?,?)', record) self.conn.commit() except sqlite3.ProgrammingError as e: print 'SQLite ERROR: ' + e.message
def CheckCursorConstructorCallCheck(self): """ Verifies that cursor methods check wether base class __init__ was called. """ class Cursor(sqlite.Cursor): def __init__(self, con): pass con = sqlite.connect(":memory:") cur = Cursor(con) try: cur.execute("select 4+5").fetchall() self.fail("should have raised ProgrammingError") except sqlite.ProgrammingError: pass except: self.fail("should have raised ProgrammingError")
def CheckConnectionConstructorCallCheck(self): """ Verifies that connection methods check wether base class __init__ was called. """ class Connection(sqlite.Connection): def __init__(self, name): pass con = Connection(":memory:") try: cur = con.cursor() self.fail("should have raised ProgrammingError") except sqlite.ProgrammingError: pass except: self.fail("should have raised ProgrammingError")
def CheckRecursiveCursorUse(self): """ http://bugs.python.org/issue10811 Recursively using a cursor, such as when reusing it from a generator led to segfaults. Now we catch recursive cursor usage and raise a ProgrammingError. """ con = sqlite.connect(":memory:") cur = con.cursor() cur.execute("create table a (bar)") cur.execute("create table b (baz)") def foo(): cur.execute("insert into a (bar) values (?)", (1,)) yield 1 with self.assertRaises(sqlite.ProgrammingError): cur.executemany("insert into b (baz) values (?)", ((i,) for i in foo()))
