Python pymysql 模块，OperationalError() 实例源码
我们从Python开源项目中，提取了以下23个代码示例，用于说明如何使用pymysql.OperationalError()。
def ExecProcQuery(self, sql):
try:
if self.connect() is None:
self.conn = None
return None
cur = self.conn.cursor()
cur.execute(sql)
resList = cur.fetchall()
self.conn.commit()
except pymysql.OperationalError as e:
if self.conn is not None:
self.conn.close()
log.e('ExecProcQuery[%s,%s]\n' % (sql, str(e)))
return None
except Exception as e:
if self.conn is not None:
self.conn.close()
log.e('ExecProcQuery[%s,%s]\n' % (sql, str(e)))
return None
return resList
def ExecProcNonQuery(self, sql):
try:
if self.connect() is None:
self.conn = None
return False
cur = self.conn.cursor()
cur.execute(sql)
self.conn.commit()
return True
except pymysql.OperationalError as e:
# self.conn.close()
if self.conn is not None:
self.conn.close()
log.e('ExecProcNonQuery[%s,%s]\n' % (sql, str(e)))
return False
except Exception as e:
if self.conn is not None:
self.conn.close()
log.e('ExecProcNonQuery[%s,%s]\n' % (sql, str(e)))
return False
def ExecNonQuery(mysql, sql):
try:
if mysql.connect() is None:
mysql.conn = None
return False
cur = mysql.conn.cursor()
cur.execute(sql)
# self.conn.commit()
return True
except pymysql.OperationalError as e:
# self.conn.close()
if mysql.conn is not None:
mysql.conn.close()
log.e('ExecProcNonQuery[%s,%s]\n' % (sql, str(e)))
return False
except Exception as e:
if mysql.conn is not None:
mysql.conn.close()
log.e('ExecProcNonQuery[%s,%s]\n' % (sql, str(e)))
return False
def find_synced_nodes(self):
"""Find a node in the cluster in SYNCED state.
:return: List of Galera node in SYNCED state.
:rtype: list(GaleraNode)
:raise: GaleraClusterSyncedNodeNotFound
"""
LOG.debug('Looking for a SYNCED node')
nodes = []
for galera_node in self._nodes:
try:
state = galera_node.wsrep_local_state
LOG.debug('%s state: %s', galera_node, state)
if state == GaleraNodeState.SYNCED:
nodes.append(galera_node)
except OperationalError as err:
LOG.error(err)
LOG.info('Skipping node %s', galera_node)
if nodes:
return nodes
else:
raise GaleraClusterSyncedNodeNotFound('Cluster has '
'no SYNCED nodes')
def realTestPamAuth(self):
db = self.db.copy()
import os
db['password'] = os.environ.get('PASSWORD')
cur = self.connections[0].cursor()
try:
cur.execute('show grants for ' + TestAuthentication.osuser + '@localhost')
grants = cur.fetchone()[0]
cur.execute('drop user ' + TestAuthentication.osuser + '@localhost')
except pymysql.OperationalError as e:
# assuming the user doesn't exist which is ok too
self.assertEqual(1045, e.args[0])
grants = None
with TempUser(cur, TestAuthentication.osuser + '@localhost',
self.databases[0]['db'], 'pam', os.environ.get('PAMSERVICE')) as u:
try:
c = pymysql.connect(user=TestAuthentication.osuser, **db)
db['password'] = 'very bad guess at password'
with self.assertRaises(pymysql.err.OperationalError):
pymysql.connect(user=TestAuthentication.osuser,
auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
**self.db)
except pymysql.OperationalError as e:
self.assertEqual(1045, e.args[0])
# we had 'bad guess at password' work with pam. Well at least we get a permission denied here
with self.assertRaises(pymysql.err.OperationalError):
pymysql.connect(user=TestAuthentication.osuser,
auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
**self.db)
if grants:
# recreate the user
cur.execute(grants)
# select old_password("crummy p\tassword");
#| old_password("crummy p\tassword") |
#| 2a01785203b08770 |
def test_issue_34(self):
try:
pymysql.connect(host="localhost", port=1237, user="root")
self.fail()
except pymysql.OperationalError as e:
self.assertEqual(2003, e.args[0])
except Exception:
self.fail()
def test_issue_35(self):
conn = self.connections[0]
c = conn.cursor()
print("sudo killall -9 mysqld within the next 10 seconds")
try:
c.execute("select sleep(10)")
self.fail()
except pymysql.OperationalError as e:
self.assertEqual(2013, e.args[0])
def test_affected_rows(self):
self.assertEqual(self.conn.affected_rows(), 0,
"Should return 0 before we do anything.")
#def test_debug(self):
## FIXME Only actually tests if you lack SUPER
#self.assertRaises(pymysql.OperationalError,
#self.conn.dump_debug_info)
def realTestPamAuth(self):
db = self.db.copy()
import os
db['password'] = os.environ.get('PASSWORD')
cur = self.connections[0].cursor()
try:
cur.execute('show grants for ' + TestAuthentication.osuser + '@localhost')
grants = cur.fetchone()[0]
cur.execute('drop user ' + TestAuthentication.osuser + '@localhost')
except pymysql.OperationalError as e:
# assuming the user doesn't exist which is ok too
self.assertEqual(1045, e.args[0])
grants = None
with TempUser(cur, TestAuthentication.osuser + '@localhost',
self.databases[0]['db'], 'pam', os.environ.get('PAMSERVICE')) as u:
try:
c = pymysql.connect(user=TestAuthentication.osuser, **db)
db['password'] = 'very bad guess at password'
with self.assertRaises(pymysql.err.OperationalError):
pymysql.connect(user=TestAuthentication.osuser,
auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
**self.db)
except pymysql.OperationalError as e:
self.assertEqual(1045, e.args[0])
# we had 'bad guess at password' work with pam. Well at least we get a permission denied here
with self.assertRaises(pymysql.err.OperationalError):
pymysql.connect(user=TestAuthentication.osuser,
auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
**self.db)
if grants:
# recreate the user
cur.execute(grants)
# select old_password("crummy p\tassword");
#| old_password("crummy p\tassword") |
#| 2a01785203b08770 |
def test_issue_34(self):
try:
pymysql.connect(host="localhost", port=1237, user="root")
self.fail()
except pymysql.OperationalError as e:
self.assertEqual(2003, e.args[0])
except Exception:
self.fail()
def test_issue_35(self):
conn = self.connections[0]
c = conn.cursor()
print("sudo killall -9 mysqld within the next 10 seconds")
try:
c.execute("select sleep(10)")
self.fail()
except pymysql.OperationalError as e:
self.assertEqual(2013, e.args[0])
def test_affected_rows(self):
self.assertEqual(self.conn.affected_rows(), 0,
"Should return 0 before we do anything.")
#def test_debug(self):
## FIXME Only actually tests if you lack SUPER
#self.assertRaises(pymysql.OperationalError,
#self.conn.dump_debug_info)
def kill_connection(self):
try:
self.Con.kill(self.Con.thread_id())
except (pymysql.OperationalError, pymysql.InternalError):
pass
def test_issue_34(self):
try:
pymysql.connect(host="localhost", port=1237, user="root")
self.fail()
except pymysql.OperationalError, e:
self.assertEqual(2003, e.args[0])
except:
self.fail()
def test_issue_35(self):
conn = self.connections[0]
c = conn.cursor()
print "sudo killall -9 mysqld within the next 10 seconds"
try:
c.execute("select sleep(10)")
self.fail()
except pymysql.OperationalError, e:
self.assertEqual(2013, e.args[0])
def test_issue_34(self):
try:
pymysql.connect(host="localhost", port=1237, user="root")
self.fail()
except pymysql.OperationalError, e:
self.assertEqual(2003, e.args[0])
except:
self.fail()
def test_issue_35(self):
conn = self.connections[0]
c = conn.cursor()
print "sudo killall -9 mysqld within the next 10 seconds"
try:
c.execute("select sleep(10)")
self.fail()
except pymysql.OperationalError, e:
self.assertEqual(2013, e.args[0])
def test_issue_34(self):
try:
pymysql.connect(host="localhost", port=1237, user="root")
self.fail()
except pymysql.OperationalError, e:
self.assertEqual(2003, e.args[0])
except:
self.fail()
def test_issue_35(self):
conn = self.connections[0]
c = conn.cursor()
print "sudo killall -9 mysqld within the next 10 seconds"
try:
c.execute("select sleep(10)")
self.fail()
except pymysql.OperationalError, e:
self.assertEqual(2013, e.args[0])
def realTestPamAuth(self):
db = self.db.copy()
import os
db['password'] = os.environ.get('PASSWORD')
cur = self.connections[0].cursor()
try:
cur.execute('show grants for ' + TestAuthentication.osuser + '@localhost')
grants = cur.fetchone()[0]
cur.execute('drop user ' + TestAuthentication.osuser + '@localhost')
except pymysql.OperationalError as e:
# assuming the user doesn't exist which is ok too
self.assertEqual(1045, e.args[0])
grants = None
with TempUser(cur, TestAuthentication.osuser + '@localhost',
self.databases[0]['db'], 'pam', os.environ.get('PAMSERVICE')) as u:
try:
c = pymysql.connect(user=TestAuthentication.osuser, **db)
db['password'] = 'very bad guess at password'
with self.assertRaises(pymysql.err.OperationalError):
pymysql.connect(user=TestAuthentication.osuser,
auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
**self.db)
except pymysql.OperationalError as e:
self.assertEqual(1045, e.args[0])
# we had 'bad guess at password' work with pam. Well at least we get a permission denied here
with self.assertRaises(pymysql.err.OperationalError):
pymysql.connect(user=TestAuthentication.osuser,
auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
**self.db)
if grants:
# recreate the user
cur.execute(grants)
# select old_password("crummy p\tassword");
#| old_password("crummy p\tassword") |
#| 2a01785203b08770 |
def test_issue_34(self):
try:
pymysql.connect(host="localhost", port=1237, user="root")
self.fail()
except pymysql.OperationalError as e:
self.assertEqual(2003, e.args[0])
except Exception:
self.fail()
def test_issue_35(self):
conn = self.connections[0]
c = conn.cursor()
print("sudo killall -9 mysqld within the next 10 seconds")
try:
c.execute("select sleep(10)")
self.fail()
except pymysql.OperationalError as e:
self.assertEqual(2013, e.args[0])