Python例外處理最佳實踐與策略

class DataAccessError(Exception):
    """針對資料存取錯誤引發的異常，例如服務中斷."""

class TransformationError(Exception):
    """當資料無法正確轉換時引發的異常."""

def retrieve_data(source):
    try:
        # 模擬資料存取錯誤，例如網路或IO故障
        raise ConnectionError("無法連線到資料來源")
    except ConnectionError as ce:
        raise DataAccessError("資料存取失敗") from ce

def transform_data(data):
    try:
        # 模擬轉換過程失敗
        if not isinstance(data, dict):
            raise ValueError("資料格式無效")
        return {"transformed": True}
    except ValueError as ve:
        raise TransformationError("資料轉換失敗") from ve

def process_pipeline(source):
    try:
        data = retrieve_data(source)
        result = transform_data(data)
    except (DataAccessError, TransformationError) as e:
        # Log the exception details at the proper level
        raise  # Re-raise after appropriate cleanup if needed

from contextlib import contextmanager

@contextmanager
def transaction_manager(resource):
    try:
        resource.begin_transaction()
        yield resource
    except Exception as ex:
        resource.rollback()
        # Optionally log or modify 'ex' to enrich error understanding
        raise
    finally:
        resource.cleanup()

# Usage in business logic
with transaction_manager(database) as db:
    db.execute("UPDATE table SET column = 'value'")

import logging
import sys

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

def log_exception(exc_type, exc_value, exc_traceback):
    if not issubclass(exc_type, KeyboardInterrupt):
        logger.error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
sys.excepthook = log_exception

try:
    # 可能出現錯誤的程式碼
    x = 1 / 0
except ZeroDivisionError:
    # 錯誤處理邏輯
    print("不能除以零!")
finally:
    # 無論是否出現錯誤都需要執行的程式碼
    print("程式結束")

try:
    x = 1 / 0
except ZeroDivisionError as e:
    raise ValueError("無效的數值") from e

class InvalidValueError(Exception):
    def __init__(self, message):
        self.message = message

try:
    x = 1 / 0
except ZeroDivisionError:
    raise InvalidValueError("無效的數值")

def access_resource(resource_id):
    try:
        # 嘗試存取資源
        data = external_api.fetch(resource_id)
    except (ConnectionError, TimeoutError) as net_exc:
        # 處理網路相關錯誤
        log.error("Network-related error occurred", exc_info=True)
        raise RuntimeError("Failed to access resource due to connectivity issues")

def calculate_statistics(data):
    try:
        # 嘗試計算統計資料
        sample = extract_sample(data)
    except (IndexError, KeyError) as extraction_error:
        # 處理資料提取錯誤
        log.exception("Extraction failed")
        raise
    else:
        # 如果沒有錯誤，則執行統計分析
        return perform_analysis(sample)

def process_file(filename):
    file = None
    try:
        # 開啟檔案
        file = open(filename, "r")
        # 處理檔案
        process(file)
    except IOError as io_err:
        # 處理檔案錯誤
        log.error("File processing error", exc_info=True)
        raise
    finally:
        # 保證檔案得到關閉
        if file is not None:
            file.close()

def process_data(data):
    try:
        # 嘗試計算統計資料
        sample = extract_sample(data)
    except (IndexError, KeyError) as extraction_error:
        # 處理資料提取錯誤
        log.exception("Extraction failed")
        raise
    else:
        # 如果沒有錯誤，則執行統計分析
        result = perform_analysis(sample)
        try:
            # 嘗試儲存結果
            save_result(result)
        except IOError as io_err:
            # 處理儲存錯誤
            log.error("Result saving error", exc_info=True)
            raise
        finally:
            # 保證資源得到釋放
            release_resources()

with open('file.txt', 'r') as file:
    content = file.read()

from contextlib import suppress

def optional_cleanup(directory):
    with suppress(FileNotFoundError):
        delete_temporary_files(directory)

import logging

def update_record(db, record_id, data):
    try:
        db.update(record_id, data)
    except Exception as exc:
        logging.error("Failed to update record %s with data %s", record_id, data, exc_info=True)
        raise

import concurrent.futures

def worker_task(task_data):
    # 處理任務資料，潛在異常必須小心處理
    process_task(task_data)

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(worker_task, data) for data in tasks]
    for future in concurrent.futures.as_completed(futures):
        try:
            future.result()
        except Exception as thread_exception:
            logging.error("Exception in worker thread", exc_info=thread_exception)

import structlog

def update_record(db, record_id, data):
    try:
        db.update(record_id, data)
    except Exception as exc:
        structlog.error("Failed to update record %s with data %s", record_id, data, exc_info=True)
        raise

import asyncio
import logging

async def async_task(i):
    # 模擬一個非同步任務
    if i < 0:
        raise ValueError("引數必須為非負數")
    return i ** 2

async def main():
    tasks = [async_task(i) for i in range(5)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            logging.error("非同步任務失敗", exc_info=result)
        else:
            logging.info("任務成功，結果為 %s", result)

asyncio.run(main())

import functools
import logging

def exception_handler(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as exc:
            logging.error("函式 %s 失敗", func.__name__, exc_info=exc)
            # 自定義處理或重新丟擲異常
            raise
    return wrapper

@exception_handler
async def critical_operation(param):
    # 複雜邏輯可能丟擲多種形式的異常
    if param < 0:
        raise ValueError("引數必須為非負數")
    return param ** 2

class ApplicationError(Exception):
    """應用程式特定的錯誤基礎類別"""
    def __init__(self, message=None, error_code=None):
        super().__init__(message)
        self.error_code = error_code