try:
    do_something()
except Exception:
    pass

# 場(chǎng)景：讀取配置文件，文件不存在時(shí)使用默認(rèn)配置
def load_config(path: str) -> dict:
    try:
        with open(path, "r") as f:
            return json.load(f)
    except FileNotFoundError:
        # 我們能處理這種情況：使用默認(rèn)值
        logger.warning(f"配置文件 {path} 不存在，使用默認(rèn)配置")
        return DEFAULT_CONFIG
    except json.JSONDecodeError as e:
        # 配置文件格式錯(cuò)誤，這是調(diào)用者的問題，重新拋出更有意義的異常
        raise ConfigurationError(f"配置文件格式錯(cuò)誤: {e}") from e

# 場(chǎng)景：數(shù)據(jù)庫(kù)寫入失敗
def save_user(user: User) -> None:
    try:
        db.execute("INSERT INTO users ...", user.to_dict())
    except DatabaseConnectionError:
        # 我無法修復(fù)數(shù)據(jù)庫(kù)連接問題，讓它向上傳播
        # 但可以加上上下文信息
        logger.error(f"保存用戶 {user.id} 失敗")
        raise  # 重新拋出原始異常，保留完整堆棧

# ? 極其危險(xiǎn)的寫法：異常被吞掉，程序悄悄出錯(cuò)
def get_user_age(user_id: int) -> int:
    try:
        user = db.get_user(user_id)
        return user.age
    except Exception:
        pass  # 發(fā)生了什么？沒人知道。

# ? 至少要記錄日志，讓問題有跡可查
def get_user_age(user_id: int) -> int | None:
    try:
        user = db.get_user(user_id)
        return user.age
    except UserNotFoundError:
        logger.warning(f"用戶 {user_id} 不存在")
        return None
    except Exception:
        logger.exception(f"獲取用戶 {user_id} 年齡時(shí)發(fā)生未預(yù)期錯(cuò)誤")
        raise  # 未知錯(cuò)誤，必須傳播

# ? 過于寬泛：一網(wǎng)打盡，隱患無窮
try:
    result = complex_calculation(data)
    save_to_database(result)
    send_notification(result)
except Exception as e:
    logger.error(f"出錯(cuò)了: {e}")

# ? 精確捕獲：每種異常獨(dú)立處理
try:
    result = complex_calculation(data)
except (ValueError, TypeError) as e:
    raise InvalidInputError(f"輸入數(shù)據(jù)格式錯(cuò)誤: {e}") from e

try:
    save_to_database(result)
except DatabaseError as e:
    logger.error(f"數(shù)據(jù)庫(kù)保存失敗，結(jié)果已緩存")
    cache.store(result)  # 降級(jí)處理
    raise

try:
    send_notification(result)
except NotificationError:
    # 通知失敗不影響主流程，記錄即可
    logger.warning("通知發(fā)送失敗，將在下次重試")

# exceptions.py —— 電商系統(tǒng)異常體系

class AppError(Exception):
    """
    應(yīng)用級(jí)基礎(chǔ)異常，所有自定義異常的根
    攜帶錯(cuò)誤碼，便于 API 響應(yīng)和監(jiān)控告警
    """
    def __init__(self, message: str, code: str = "APP_ERROR", details: dict = None):
        super().__init__(message)
        self.message = message
        self.code = code
        self.details = details or {}

    def to_dict(self) -> dict:
        """轉(zhuǎn)換為 API 響應(yīng)格式"""
        return {
            "error": self.code,
            "message": self.message,
            "details": self.details
        }

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(code={self.code!r}, message={self.message!r})"


# ─── 領(lǐng)域?qū)樱喊礃I(yè)務(wù)模塊劃分 ────────────────────────

class UserError(AppError):
    """用戶模塊異?；?""
    pass

class UserNotFoundError(UserError):
    """用戶不存在"""
    def __init__(self, user_id: int | str):
        super().__init__(
            message=f"用戶 {user_id} 不存在",
            code="USER_NOT_FOUND",
            details={"user_id": user_id}
        )
        self.user_id = user_id

class UserPermissionError(UserError):
    """用戶權(quán)限不足"""
    def __init__(self, user_id: int, required_permission: str):
        super().__init__(
            message=f"用戶 {user_id} 缺少權(quán)限: {required_permission}",
            code="PERMISSION_DENIED",
            details={"user_id": user_id, "required": required_permission}
        )


class OrderError(AppError):
    """訂單模塊異?；?""
    pass

class OrderNotFoundError(OrderError):
    def __init__(self, order_id: str):
        super().__init__(
            message=f"訂單 {order_id} 不存在",
            code="ORDER_NOT_FOUND",
            details={"order_id": order_id}
        )

class InsufficientStockError(OrderError):
    """庫(kù)存不足"""
    def __init__(self, product_id: str, requested: int, available: int):
        super().__init__(
            message=f"商品 {product_id} 庫(kù)存不足（需要 {requested}，剩余 {available}）",
            code="INSUFFICIENT_STOCK",
            details={
                "product_id": product_id,
                "requested": requested,
                "available": available
            }
        )

class OrderStateError(OrderError):
    """訂單狀態(tài)流轉(zhuǎn)錯(cuò)誤"""
    def __init__(self, order_id: str, current_state: str, expected_state: str):
        super().__init__(
            message=f"訂單 {order_id} 當(dāng)前狀態(tài)為 {current_state}，無法執(zhí)行此操作（需要 {expected_state}）",
            code="INVALID_ORDER_STATE",
            details={
                "order_id": order_id,
                "current_state": current_state,
                "expected_state": expected_state
            }
        )


# ─── 基礎(chǔ)設(shè)施層：技術(shù)異常 ────────────────────────

class InfrastructureError(AppError):
    """基礎(chǔ)設(shè)施層異?；?""
    pass

class DatabaseError(InfrastructureError):
    def __init__(self, operation: str, cause: Exception = None):
        super().__init__(
            message=f"數(shù)據(jù)庫(kù)操作失敗: {operation}",
            code="DATABASE_ERROR"
        )
        self.__cause__ = cause

class ExternalServiceError(InfrastructureError):
    """第三方服務(wù)調(diào)用失敗"""
    def __init__(self, service_name: str, status_code: int = None):
        super().__init__(
            message=f"外部服務(wù) {service_name} 調(diào)用失敗",
            code="EXTERNAL_SERVICE_ERROR",
            details={"service": service_name, "status_code": status_code}
        )

# order_service.py —— 異常體系的實(shí)際使用

class OrderService:

    def create_order(self, user_id: int, items: list[dict]) -> Order:
        """創(chuàng)建訂單，展示完整的異常處理鏈路"""

        # 1. 驗(yàn)證用戶
        user = self._get_user_or_raise(user_id)

        # 2. 檢查庫(kù)存（精確捕獲，各個(gè)擊破）
        for item in items:
            self._check_stock(item["product_id"], item["quantity"])

        # 3. 創(chuàng)建訂單
        try:
            order = Order.create(user_id=user_id, items=items)
            self.db.save(order)
            return order
        except Exception as e:
            # 將底層異常轉(zhuǎn)換為業(yè)務(wù)異常，附加上下文
            raise DatabaseError("創(chuàng)建訂單", cause=e) from e

    def _get_user_or_raise(self, user_id: int) -> User:
        """獲取用戶，不存在則拋出語義明確的異常"""
        user = self.db.find_user(user_id)
        if user is None:
            raise UserNotFoundError(user_id)
        return user

    def _check_stock(self, product_id: str, quantity: int) -> None:
        """檢查庫(kù)存，不足則拋出攜帶詳細(xì)信息的異常"""
        product = self.db.find_product(product_id)
        if product.stock < quantity:
            raise InsufficientStockError(
                product_id=product_id,
                requested=quantity,
                available=product.stock
            )

    def cancel_order(self, order_id: str, user_id: int) -> None:
        """取消訂單，演示狀態(tài)校驗(yàn)異常"""
        order = self.db.find_order(order_id)
        if order is None:
            raise OrderNotFoundError(order_id)

        if order.status != "pending":
            raise OrderStateError(
                order_id=order_id,
                current_state=order.status,
                expected_state="pending"
            )

        order.cancel()
        self.db.save(order)

# api/exception_handlers.py —— FastAPI 統(tǒng)一異常處理

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    """
    統(tǒng)一處理所有業(yè)務(wù)異常，轉(zhuǎn)換為標(biāo)準(zhǔn) HTTP 響應(yīng)
    """
    # 根據(jù)異常類型決定 HTTP 狀態(tài)碼
    status_code_map = {
        "USER_NOT_FOUND": 404,
        "ORDER_NOT_FOUND": 404,
        "PERMISSION_DENIED": 403,
        "INSUFFICIENT_STOCK": 409,
        "INVALID_ORDER_STATE": 422,
        "DATABASE_ERROR": 503,
        "EXTERNAL_SERVICE_ERROR": 502,
    }
    status_code = status_code_map.get(exc.code, 500)

    # 服務(wù)端錯(cuò)誤記錄詳細(xì)日志
    if status_code >= 500:
        logger.error(f"服務(wù)端錯(cuò)誤: {exc!r}", exc_info=True)
    else:
        logger.info(f"業(yè)務(wù)異常: {exc!r}")

    return JSONResponse(
        status_code=status_code,
        content=exc.to_dict()
    )

@app.exception_handler(Exception)
async def unhandled_error_handler(request: Request, exc: Exception) -> JSONResponse:
    """兜底處理：未被捕獲的異常"""
    logger.critical(f"未處理的異常: {exc!r}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"error": "INTERNAL_ERROR", "message": "服務(wù)器內(nèi)部錯(cuò)誤"}
    )

# ? 異常鏈：保留原始原因，同時(shí)提供業(yè)務(wù)上下文
try:
    raw_data = json.loads(request_body)
except json.JSONDecodeError as e:
    raise InvalidRequestError("請(qǐng)求體不是合法的 JSON 格式") from e
    # 上面這行讓異常信息變?yōu)椋?
    # InvalidRequestError: 請(qǐng)求體不是合法的 JSON 格式
    # The above exception was the direct cause of the following exception:
    # json.JSONDecodeError: ...（原始錯(cuò)誤保留）

# ? 丟失原始異常信息（調(diào)試時(shí)抓瞎）
try:
    raw_data = json.loads(request_body)
except json.JSONDecodeError:
    raise InvalidRequestError("請(qǐng)求體不是合法的 JSON 格式")

# 自定義上下文管理器：事務(wù)管理
from contextlib import contextmanager

@contextmanager
def db_transaction(db_session):
    """
    確保數(shù)據(jù)庫(kù)事務(wù)在異常時(shí)自動(dòng)回滾
    """
    try:
        yield db_session
        db_session.commit()
        logger.debug("事務(wù)提交成功")
    except AppError:
        db_session.rollback()
        logger.warning("業(yè)務(wù)異常，事務(wù)已回滾")
        raise  # 業(yè)務(wù)異常繼續(xù)傳播
    except Exception as e:
        db_session.rollback()
        logger.error("未知異常，事務(wù)已回滾", exc_info=True)
        raise DatabaseError("事務(wù)執(zhí)行失敗") from e
    finally:
        db_session.close()

# 使用
with db_transaction(session) as txn:
    txn.execute("UPDATE ...")
    txn.execute("INSERT ...")
    # 任何異常都會(huì)觸發(fā)回滾

import time
import functools
from typing import Type

def retry(
    exceptions: tuple[Type[Exception], ...],
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0
):
    """
    裝飾器：對(duì)指定異常類型進(jìn)行自動(dòng)重試（指數(shù)退避）
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempt = 0
            current_delay = delay

            while attempt < max_attempts:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    attempt += 1
                    if attempt >= max_attempts:
                        logger.error(f"{func.__name__} 重試 {max_attempts} 次后仍失敗: {e}")
                        raise

                    logger.warning(
                        f"{func.__name__} 第 {attempt} 次失敗: {e}，"
                        f"{current_delay:.1f}s 后重試"
                    )
                    time.sleep(current_delay)
                    current_delay *= backoff

        return wrapper
    return decorator


# 使用：對(duì)網(wǎng)絡(luò)請(qǐng)求、外部服務(wù)調(diào)用啟用重試
@retry(exceptions=(ExternalServiceError, ConnectionError), max_attempts=3, delay=0.5)
def call_payment_api(order_id: str, amount: float) -> dict:
    response = requests.post(PAYMENT_API_URL, json={"order_id": order_id, "amount": amount})
    if response.status_code != 200:
        raise ExternalServiceError("payment-api", response.status_code)
    return response.json()

# 反模式一：用異?？刂普Ａ鞒蹋ㄐ阅懿?，語義混亂）
# ?
def find_user(user_id):
    try:
        return User.objects.get(id=user_id)
    except User.DoesNotExist:
        return None  # 這種情況應(yīng)該用 .filter().first()

# ?
def find_user(user_id):
    return User.objects.filter(id=user_id).first()


# 反模式二：過于細(xì)碎的 try/except 塊
# ?
try:
    a = int(input_a)
except ValueError:
    a = 0
try:
    b = int(input_b)
except ValueError:
    b = 0

# ? 提取成函數(shù)
def safe_int(value: str, default: int = 0) -> int:
    try:
        return int(value)
    except (ValueError, TypeError):
        return default

a = safe_int(input_a)
b = safe_int(input_b)


# 反模式三：在 finally 中使用 return（會(huì)吞掉異常?。?
# ?
def dangerous():
    try:
        raise ValueError("出錯(cuò)了")
    finally:
        return 42  # 異常被吞掉，函數(shù)返回 42

# ? finally 只做清理，不做返回
def safe():
    try:
        raise ValueError("出錯(cuò)了")
    finally:
        cleanup()  # 只清理資源

# Python 3.11+：并發(fā)任務(wù)的多異常處理
async def fetch_all(urls):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url)) for url in urls]

# 使用 except* 捕獲特定類型的并發(fā)異常
try:
    await fetch_all(urls)
except* TimeoutError as eg:
    print(f"超時(shí)的任務(wù)數(shù): {len(eg.exceptions)}")
except* ConnectionError as eg:
    print(f"連接失敗的任務(wù)數(shù): {len(eg.exceptions)}")

from dataclasses import dataclass
from typing import Generic, TypeVar

T = TypeVar("T")
E = TypeVar("E", bound=Exception)

@dataclass
class Ok(Generic[T]):
    value: T

@dataclass
class Err(Generic[E]):
    error: E

Result = Ok[T] | Err[E]

def safe_divide(a: float, b: float) -> Result:
    if b == 0:
        return Err(ZeroDivisionError("除數(shù)不能為零"))
    return Ok(a / b)

# 調(diào)用者顯式處理兩種情況
match safe_divide(10, 0):
    case Ok(value=v):
        print(f"結(jié)果: {v}")
    case Err(error=e):
        print(f"計(jì)算失敗: {e}")