Python_學習之上下文

目錄

示例1：查看上下文執行的順序
示例2：動態控制上下文是否拋出異常
示例3：以裝飾器的方式爲功能函數加裝上下文
示例4：過濾異常，不拋出

在以前咱們進行過文件操做的學習時，咱們爲了避免忘掉文件操做完畢後關閉文件file.close()，官方推薦推薦咱們使用with……as 語句，這其實本質就是運用了python的上下文管理。

而所謂的上下文，其實就是服務運行的狀態從進入到退出的一種過程，python中咱們經常經過上下文來進行資源的建立與釋放。

語法：with……as

本質：

程序執行with中的代碼時，會自動先執行enter方法，返回在這個上下文中使用的對象句柄，程序執行完邏輯後自動調用exit來進行資源的釋放

示例1：查看上下文執行的順序
若是上下文中出現異常即執行邏輯代碼過程當中，上下文是能夠捕獲異常的，而且默認是拋出異常的

class MyContext:
   def __init__(self):
       print("in __init__")

   def __enter__(self):
       print("int __enter__")
       return self

   def __exit__(self, exc_type, exc_val, exc_tb):
       """"""
       print("in __exit__")
       print("異常的類型exc_type=", exc_type)
       print("異常拋出的值exc_type=", exc_val)
       print("異常的traceback對象exc_type=", exc_tb)

if __name__ == '__main__':
   with MyContext() as t:
       print("執行代碼邏輯……")
       raise Exception("錯誤解釋")

# 執行結果爲
"""
in __init__
int __enter__
執行代碼邏輯……
in __exit__
異常的類型exc_type= <class 'Exception'>
異常拋出的值exc_type= 錯誤解釋
異常的traceback對象exc_type= <traceback object at 0x000001B52B5465C8>
Traceback (most recent call last):
File "D:/my_all_project/Frame_learning/test.py", line 27, in <module>
  raise Exception("錯誤解釋")
Exception: 錯誤解釋
"""

示例2：動態控制上下文是否拋出異常
若是功能函數邏輯中出現異常，而exit方法返回值等價於False就會拋出異常，不然不拋出異常，繼續執行上下文外面的業務邏輯

class MyContext:
   def __init__(self, flag):
       print("in __init__")
       self.flag = flag

   def __enter__(self):
       print("int __enter__")
       "能夠返回咱們定義任何方法或實例化的對象"
       return self

   def __exit__(self, exc_type, exc_val, exc_tb):
       """"""
       print("in __exit__")
       print("異常的類型exc_type=", exc_type)
       print("異常拋出的值exc_type=", exc_val)
       print("異常的traceback對象exc_type=", exc_tb)
       return self.flag

if __name__ == '__main__':
   with MyContext(True) as t:
       print("執行代碼邏輯……")
       raise Exception("錯誤解釋")
   print("===>當上下文不拋出異常時，此處能夠被打印")

# t 實際就是類Mycontext的實例化對象，其能夠調用類中的任意實例方法和屬性

示例3：以裝飾器的方式爲功能函數加裝上下文

import contextlib

class MyContext(contextlib.ContextDecorator):

   def __init__(self):
       print(f'__init__()')

   def __enter__(self):
       print(f'__enter__()')
       return self

   def __exit__(self, exc_type, exc_val, exc_tb):
       print(f'__exit__()')
       print(f"exc_type:{exc_type}")
       try:
           self.write_log()
       except Exception as e:
           print(e)

   @staticmethod
   def write_log():
       print("開始記錄日誌")
       log = dict(req_body="request", rsp_body="response")
       f = open("my_log_test.txt", "w", encoding="utf-8")
       import json
       f.write(json.dumps(log, ensure_ascii=False))
       f.flush()
       f.close()
       raise Exception("本日誌記錄報錯，不在影響功能函數的正常返回")

@MyContext()
def func(flag):
   code, desc = 1, "fail"
   try:
       if flag:
           raise Exception("測試上下文是否能捕獲異常")
       else:
           code, desc = 0, "success"
   except Exception as e:
       print(f"本初捕獲異常:{e},則不會再拋給上下文管理器中")
   else:
       code, desc = 0, "success"
   finally:
       return {"code": code, "desc": desc}

if __name__ == '__main__':
   ret = func(True)
   print(ret)

# 本小例是經過上下文初試爲功能函數添加記錄日誌的功能，不因記錄日誌出現異常致使功能函數異常，也能夠加一個開關，是否記錄日誌

# 寫日誌的另外一個版本
import threading

class WriteLogContext:
   def __init__(self, flag, data):
       """
      :param flag: 異常日誌是否拋出開關標識
      :param data: 日誌內容
      """
       self.flag = flag
       self.data = data

   def __enter__(self):
       return self

   def __exit__(self, exc_type, exc_val, exc_tb):
       print("異常的類型exc_type=", exc_type)
       print("異常拋出的值exc_type=", exc_val)
       print("異常的traceback對象exc_type=", exc_tb)
       return self.flag

   def write_action(self):
       """開啓多線程記錄日誌,發現異常不會拋出外層，但會將異常打印到控制檯"""
       write_external_thread = threading.Thread(target=self.write_log, args=(self.data,))
       write_external_thread.setDaemon(True)
       write_external_thread.start()

   @staticmethod
   def write_log(log):
       with open("test.txt", "a") as file:
           file.write("入庫的操做模擬\n")
           file.write(f"{log}\n")
           file.flush()
       # 模擬異常
       raise TypeError("模擬寫日誌過程當中的異常,發現本處報錯，並不會影響主功能函數響應")

def access():
   # 執行業務
   print("business is begin")
   # 記錄流水
   log = "life is short ,i use python ,i use it for make money"
   with WriteLogContext(flag=True, data=log) as f:
       f.write_action()
   # 響應
   result = "business is success"
   return result

示例4：過濾異常，不拋出

import contextlib

def write_log(data):
   if data:
       print(111111)
   else:
       raise Exception("life")

with contextlib.suppress(Exception):
   data = dict()
   write_log(data)

# suppress類能夠過濾指定的異常，不拋出