Python実務の型：例外処理と構造化ログでエラーに強いコードを書く

class AppError(Exception):
    """アプリ共通の基底例外（捕捉はこれ1つでOK）"""

class ExternalError(AppError):
pass  # API/ネット/DB 由来

class DataError(AppError):
pass  # パース/検証 由来

class BusinessRuleError(AppError):
pass  # 業務ルール違反

from pathlib import Path
import json

def load\_config(path: Path) -> dict:
try:
text = path.read\_text(encoding="utf-8")
return json.loads(text)
except FileNotFoundError as e:
raise ExternalError(f"config not found: {path}") from e
except json.JSONDecodeError as e:
raise DataError(f"invalid json: {path}: {e}") from e

f = open("data.csv", "r", encoding="utf-8")
try:
    rows = f.readlines()     # 例外が出得る本処理
except OSError as e:
    ...                      # ログ＆代替
else:
    ...                      # 成功時だけ実行
finally:
    f.close()                # 成否に関わらず必ず実行
# 実務では with 文が推奨

import time, random, urllib.request, logging

log = logging.getLogger(**name**)

def get\_with\_retry(url: str, \*, tries: int = 5, base: float = 0.5, cap: float = 8.0) -> str:
"""指数バックオフ＋フルジッタ。最大 tries で中断。"""
for attempt in range(1, tries + 1):
try:
with urllib.request.urlopen(url, timeout=5) as r:
if r.status >= 500:
raise ExternalError(f"server {r.status}")
return r.read().decode()
except (TimeoutError, OSError) as e:
\# リトライ対象：ネット系の一過性エラー
if attempt == tries:
raise ExternalError(f"retry exceeded: {url}") from e
sleep = min(cap, base \* 2 \*\* (attempt - 1)) \* random.random()
log.warning("retry %s/%s in %.2fs: %s", attempt, tries, sleep, e)
time.sleep(sleep)

import logging, logging.config, json
from logging.handlers import TimedRotatingFileHandler

class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
payload = {
"ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
"lvl": record.levelname,
"msg": record.getMessage(),
"logger": record.name,
"module": record.module,
"line": record.lineno,
"corr": getattr(record, "corr", "-"),
}
if record.exc\_info:
payload\["exc"] = self.formatException(record.exc\_info)
return json.dumps(payload, ensure\_ascii=False)

handler = TimedRotatingFileHandler("app.log", when="midnight", backupCount=7, encoding="utf-8")
handler.setFormatter(JsonFormatter())

root = logging.getLogger()
root.setLevel(logging.INFO)
root.addHandler(handler)

import logging, uuid, contextvars

corr\_var: contextvars.ContextVar\[str] = contextvars.ContextVar("corr", default="-")

class CorrAdapter(logging.LoggerAdapter):
def process(self, msg, kwargs):
extra = kwargs.setdefault("extra", {})
extra\["corr"] = corr\_var.get()
return msg, kwargs

base = logging.getLogger(**name**)
log = CorrAdapter(base, {})

def with\_corr(func):
def wrapper(\*a, \*\*kw):
token = corr\_var.set(uuid.uuid4().hex\[:12])
try:
return func(\*a, \*\*kw)
finally:
corr\_var.reset(token)
return wrapper

@with\_corr
def run\_job():
log.info("start job")
try:
...
except AppError:
log.exception("failed")  # exc\_info 付き
raise
finally:
log.info("end job")

try:
    risky()
except Exception:
    pass  # 何もせず無視

try:
    risky()
except DataError as e:
    log.warning("invalid data: %s", e, exc_info=True)
    return Fallback()

# app.py
class AppError(Exception): ...
class DataError(AppError): ...

def parse\_int(s: str) -> int:
try:
return int(s)
except ValueError as e:
raise DataError(f"not int: {s}") from e

# tests/test_app.py
import pytest
from app import parse_int, DataError

def test\_parse\_int\_ok():
assert parse\_int("42") == 42

def test\_parse\_int\_ng():
with pytest.raises(DataError) as ei:
parse\_int("xx")
assert "not int" in str(ei.value)