8.1.1 處理警告

與執行單一陳述式類似,認可或回滾交易也會觸發警告。為了能夠處理這些警告,需要檢查 Session.commit();Session.rollback(); 的回覆結果物件。

以下範例顯示了這一點。此範例假設測試綱要存在,且集合 my_collection 不存在。

from mysqlsh import mysqlx

# Connect to server
mySession = mysqlx.get_session( {
        'host': 'localhost', 'port': 33060,
        'user': 'user', 'password': 'password' } )

# Get the Schema test
myDb = mySession.get_schema('test')

# Create a new collection
myColl = myDb.create_collection('my_collection')

# Start a transaction
mySession.start_transaction()
try:
    myColl.add({'name': 'Rohit', 'age': 18, 'height': 1.76}).execute()
    myColl.add({'name': 'Misaki', 'age': 24, 'height': 1.65}).execute()
    myColl.add({'name': 'Leon', 'age': 39, 'height': 1.9}).execute()

    # Commit the transaction if everything went well
    reply = mySession.commit()

    # handle warnings
    if reply.warning_count:
      for warning in result.get_warnings():
        print('Type [%s] (Code %s): %s\n' % (warning.level, warning.code, warning.message))

    print('Data inserted successfully.')
except Exception as err:
    # Rollback the transaction in case of an error
    reply = mySession.rollback()

    # handle warnings
    if reply.warning_count:
      for warning in result.get_warnings():
        print('Type [%s] (Code %s): %s\n' % (warning.level, warning.code, warning.message))

    # Printing the error message
    print('Data could not be inserted: %s' % str(err))

預設情況下,所有警告都會從伺服器傳送到用戶端。如果已知某項操作會產生許多警告,且這些警告對應用程式沒有任何價值,則可以抑制傳送警告。這有助於節省頻寬。session.setFetchWarnings() 控制是否在伺服器上捨棄警告,或是否傳送到用戶端。session.getFetchWarnings() 用於了解目前作用中的設定。

from mysqlsh import mysqlx

def process_warnings(result):
  if result.get_warnings_count():
    for warning in result.get_warnings():
      print('Type [%s] (Code %s): %s\n' % (warning.level, warning.code, warning.message))
  else:
    print("No warnings were returned.\n")


# Connect to server
mySession = mysqlx.get_session( {
  'host': 'localhost', 'port': 33060,
  'user': 'user', 'password': 'password' } );

# Disables warning generation
mySession.set_fetch_warnings(False)
result = mySession.sql('drop schema if exists unexisting').execute()
process_warnings(result)

# Enables warning generation
mySession.set_fetch_warnings(True)
result = mySession.sql('drop schema if exists unexisting').execute()
process_warnings(result)