Coverage for config / settings.py: 97.60%

206 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-06-17 01:43 +0000

1"""アプリケーション設定管理 

2 

3学習目標: 

4- Pydantic Settingsを使った環境設定管理 

5- 環境ごとの設定分離パターン 

6- 設定値のバリデーション 

7- セキュリティベストプラクティス 

8""" 

9 

10import ipaddress 

11import logging 

12import os 

13import socket 

14from enum import StrEnum 

15from functools import lru_cache 

16from pathlib import Path 

17from typing import Any, Self 

18from urllib.parse import urlparse 

19 

20from pydantic import BaseModel, Field, SecretStr, field_validator, model_validator 

21from pydantic_settings import BaseSettings, SettingsConfigDict 

22 

23# NOTE: utils/logger.py は config.settings に依存するため structlog は使用不可(循環インポート回避) 

24_logger = logging.getLogger(__name__) 

25 

26# ============================================================================= 

27# SSRF Prevention Configuration 

28# ============================================================================= 

29 

30 

31def _get_allowed_domains() -> frozenset[str]: 

32 """環境変数またはデフォルト値から許可ドメインリストを取得 

33 

34 環境変数 ALLOWED_DOMAINS: カンマ区切りのドメインリスト 

35 例: ALLOWED_DOMAINS=api.example.com,api.test.com 

36 空白のみの値は空集合を返す(deny-all)。 

37 """ 

38 if "ALLOWED_DOMAINS" in os.environ: 

39 env_domains = os.environ["ALLOWED_DOMAINS"] 

40 # 余分な空白は除去する。空文字列・空白のみなら空集合になり、deny-all を維持する。 

41 domains = frozenset(d.strip() for d in env_domains.split(",") if d.strip()) 

42 if not domains: 

43 # deny-all (全ドメイン拒否) は設定ミスの可能性が高いため起動時に警告する。 

44 # 空の許可リストは全 validate_base_url() を失敗させるため、サイレントな 

45 # 許可リスト空化 (例: .env の `ALLOWED_DOMAINS=`) を早期検出可能にする。 

46 _logger.warning( 

47 "ALLOWED_DOMAINS が空のため全ドメインを拒否します (deny-all)。" 

48 " SSRF 許可リストが意図せず空になっていないか確認してください。 raw value=%r", 

49 env_domains[:200], 

50 ) 

51 return domains 

52 

53 # デフォルト: 本番用 + テスト用ドメイン 

54 return frozenset( 

55 { 

56 # 本番用 

57 "jsonplaceholder.typicode.com", 

58 "api.github.com", 

59 "httpbin.org", 

60 # テスト用(example.com は RFC 2606 で予約済み) 

61 "example.com", 

62 "api.example.com", 

63 "test.example.com", 

64 "test-api.example.com", 

65 } 

66 ) 

67 

68 

69# 許可されたドメインリスト(環境変数で上書き可能) 

70# NOTE: モジュール読み込み時に確定する。起動後の環境変数変更(monkeypatch等)は 

71# 再起動するまで validate_base_url() に反映されない。validate_base_url() は 

72# モジュール変数 ALLOWED_DOMAINS を直接参照するため、テストで動的に変更する場合は 

73# config.settings.ALLOWED_DOMAINS を monkeypatch するか、モジュールインポート前に 

74# 環境変数を設定すること。_get_allowed_domains() を呼ぶだけでは戻り値が 

75# ALLOWED_DOMAINS に再代入されず validate_base_url() の動作は変わらない。 

76# ALLOWED_DOMAINS は module-level で評価されるため、settings インスタンス生成タイミングではなく 

77# import 前の環境変数設定が必要。 

78ALLOWED_DOMAINS: frozenset[str] = _get_allowed_domains() 

79 

80 

81# 危険なプライベートIPレンジ 

82PRIVATE_IP_RANGES: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = ( 

83 ipaddress.ip_network("10.0.0.0/8"), 

84 ipaddress.ip_network("172.16.0.0/12"), 

85 ipaddress.ip_network("192.168.0.0/16"), 

86 ipaddress.ip_network("127.0.0.0/8"), # loopback 

87 ipaddress.ip_network("169.254.0.0/16"), # link-local (AWS metadata) 

88 ipaddress.ip_network("0.0.0.0/8"), # current network / non-routable 

89 ipaddress.ip_network("100.64.0.0/10"), # carrier-grade NAT 

90 ipaddress.ip_network("::1/128"), # IPv6 loopback 

91 ipaddress.ip_network("fc00::/7"), # IPv6 private 

92 ipaddress.ip_network("fe80::/10"), # IPv6 link-local 

93) 

94 

95 

96def _check_ip_private(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: 

97 """IPアドレスがプライベートかチェック(IPv4-mapped IPv6対応) 

98 

99 Security: 

100 IPv4-mapped IPv6(::ffff:x.x.x.x)を使ったSSRFバイパス攻撃を防止。 

101 例: ::ffff:192.168.1.1 は実質的に192.168.1.1と同じ。 

102 """ 

103 # IPv4-mapped IPv6アドレスの検出と変換 

104 if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped: 

105 # ::ffff:192.168.1.1 → 192.168.1.1 として評価 

106 ip = ip.ipv4_mapped 

107 

108 return any(ip in network for network in PRIVATE_IP_RANGES) 

109 

110 

111# 起動時/設定検証の短期再利用を想定したプロセス内キャッシュ。 

112# 長期稼働中のDNS変更追従が必要な用途ではTTL付きキャッシュへ切り替える。 

113@lru_cache(maxsize=256) 

114def _resolve_hostname_cached(hostname: str) -> str: 

115 """ホスト名をDNS解決してIPアドレス文字列を返す(成功時のみキャッシュ) 

116 

117 Args: 

118 hostname: 解決対象のホスト名 

119 

120 Returns: 

121 解決されたIPアドレス文字列。 

122 

123 Raises: 

124 socket.herror: DNSサーバーエラーによる解決失敗。 

125 socket.gaierror: アドレス情報取得失敗。 

126 OSError: socket.herror / socket.gaierror を含む全ネットワークエラーの基底クラス。 

127 ここでは PermissionError / TimeoutError 等それ以外の OSError サブクラスを指す。 

128 UnicodeError: 非ASCII文字を含むホスト名等のUnicode処理エラー時。 

129 TypeError: NULバイト(\\x00)等の不正文字を含むホスト名の場合。 

130 OverflowError: 極端に長いホスト名によるOSレベルオーバーフロー時(プラットフォーム依存)。 

131 """ 

132 return socket.gethostbyname(hostname) 

133 

134 

135def _resolve_hostname(hostname: str) -> str | None: 

136 """ホスト名をDNS解決してIPアドレス文字列を返す(失敗時はNone) 

137 

138 Args: 

139 hostname: 解決対象のホスト名 

140 

141 Returns: 

142 解決されたIPアドレス文字列。失敗時はNone(キャッシュされない)。 

143 

144 Note: 

145 DNS解決成功時のみ _resolve_hostname_cached でキャッシュされる。 

146 一時的なDNS障害後は次の呼び出しで再試行が行われる。 

147 

148 Security: 

149 不正ホスト名入力に対するSSRF防止: 

150 - UnicodeDecodeError: DNS応答の不正バイト列デコード失敗 

151 - UnicodeEncodeError: 非ASCII文字を含むホスト名のエンコード失敗 

152 - UnicodeError(親クラス)でまとめて捕捉し、実装・プラットフォーム差を吸収 

153 - TypeError: NULバイト(\x00)等の不正文字を含むホスト名 

154 - OverflowError: 極端に長いホスト名によるOSレベルオーバーフロー(プラットフォーム依存) 

155 - ラッパー側でcatchすることでlru_cacheに影響せず(成功値のみキャッシュ維持) 

156 - Fail-Closed: Noneを返してis_private_ipがブロック判定する 

157 """ 

158 try: 

159 return _resolve_hostname_cached(hostname) 

160 except (UnicodeError, OverflowError, TypeError) as e: 

161 # UnicodeError: Unicode処理エラー(UnicodeDecodeError/UnicodeEncodeError含む) 

162 # - 非ASCII文字を含む攻撃的なホスト名(SSRF試行の可能性) 

163 # OverflowError: 極端に長いホスト名によるOSレベルオーバーフロー(プラットフォーム依存) 

164 # TypeError: NULバイト(\x00)等の不正文字を含むホスト名 

165 _logger.warning( 

166 "不正なホスト名形式 — SSRF試行の可能性: hostname=%r, error_type=%s, error=%r", 

167 hostname[:200], 

168 type(e).__name__, 

169 e, 

170 ) 

171 return None 

172 except (socket.herror, socket.gaierror) as e: 

173 # DNS解決失敗(herror: サーバーエラー, gaierror: アドレス情報エラー) 

174 # Fail-Closed: サービスへの影響(正当リクエストのブロック)が発生するためWARNINGレベル 

175 _logger.warning( 

176 "DNS解決失敗: hostname=%r — ブロック扱い (error_type=%s, errno=%s)", 

177 hostname[:200], 

178 type(e).__name__, 

179 e.args[0] if e.args else "N/A", 

180 ) 

181 return None 

182 except OSError as e: 

183 # 予期しないネットワークエラー(PermissionError, TimeoutError 等) 

184 # Fail-Closed: DNS解決失敗として誤分類しないよう別メッセージで記録 

185 _logger.warning( 

186 "予期しないネットワークエラー: hostname=%r — ブロック扱い (error_type=%s, errno=%s)", 

187 hostname[:200], 

188 type(e).__name__, 

189 getattr(e, "errno", "N/A"), 

190 ) 

191 return None 

192 

193 

194def is_private_ip(hostname: str) -> bool: 

195 """ホスト名がプライベートIPまたはローカルアドレスかチェック 

196 

197 Args: 

198 hostname: チェック対象のホスト名またはIPアドレス 

199 

200 Returns: 

201 True: プライベート/ローカルIP, False: パブリックIP 

202 

203 Security: 

204 - IPv4-mapped IPv6(::ffff:x.x.x.x)もプライベートIPとして検出 

205 - DNS解決失敗時はFail-Closed(ブロック) 

206 - DNS解決結果はLRUキャッシュ(256エントリ)で高速化 

207 

208 """ 

209 try: 

210 # IPアドレス形式の場合(DNS解決不要) 

211 ip = ipaddress.ip_address(hostname) 

212 return _check_ip_private(ip) 

213 except ValueError: 

214 # ホスト名の場合、DNS解決を試みる(キャッシュ付き) 

215 resolved = _resolve_hostname(hostname) 

216 if resolved is None: 

217 # DNS解決失敗は安全側に倒す(ブロック = Fail-Closed) 

218 # セキュリティ: SSRF攻撃防止のため、不明なホストはプライベートIPと見なす 

219 return True 

220 try: 

221 ip = ipaddress.ip_address(resolved) 

222 return _check_ip_private(ip) 

223 except ValueError: 

224 # DNS解決結果が有効なIPアドレス形式でない場合(異常なDNS応答) 

225 # セキュリティ: 不正な値はプライベートIPとして扱いブロック(Fail-Closed) 

226 # ログを残すことでセキュリティインシデントの証拠を保全する 

227 # _logger はモジュールレベル(インポート直後)で定義済み 

228 _logger.warning( 

229 "DNS解決結果が不正なIPアドレス形式: hostname=%r, resolved=%r — ブロック扱い", 

230 hostname[:200], 

231 resolved[:100], 

232 ) 

233 return True 

234 

235 

236def _validate_base_url_with_allowed_domains(v: str, allowed_domains: frozenset[str]) -> str: 

237 """base_url 検証を切り出す。 

238 

239 許可ドメインを注入できるようにして、Settings の全体初期化に依存しない 

240 直接テストを可能にする。 

241 """ 

242 if not v.startswith(("http://", "https://")): 

243 raise ValueError("Base URL must start with http:// or https://") 

244 

245 # URLパース 

246 parsed = urlparse(v) 

247 hostname = parsed.hostname 

248 

249 if not hostname: 

250 raise ValueError("Invalid URL: hostname not found") 

251 

252 # SSRF Prevention: プライベートIPチェック 

253 if is_private_ip(hostname): 

254 _logger.warning( 

255 "SSRF Prevention: private or loopback IP blocked for hostname=%r. " 

256 "If this is a valid hostname, check DNS resolution and ALLOWED_DOMAINS setting.", 

257 hostname[:200], 

258 ) 

259 raise ValueError("SSRF Prevention: Private/loopback IP addresses are not allowed.") 

260 

261 # SSRF Prevention: 許可ドメインチェック 

262 if hostname not in allowed_domains: 

263 _logger.warning( 

264 "SSRF Prevention: Domain not in allowlist: %r. Allowed domains count: %d. " 

265 "Check ALLOWED_DOMAINS setting.", 

266 hostname[:200], 

267 len(allowed_domains), 

268 ) 

269 raise ValueError("SSRF Prevention: Domain not in allowlist.") 

270 

271 return v.rstrip("/") 

272 

273 

274# ============================================================================= 

275# 環境定義 

276# ============================================================================= 

277 

278 

279class Environment(StrEnum): 

280 """実行環境の定義""" 

281 

282 DEVELOPMENT = "development" 

283 TESTING = "testing" 

284 STAGING = "staging" 

285 PRODUCTION = "production" 

286 

287 

288class LogLevel(StrEnum): 

289 """ログレベルの定義""" 

290 

291 DEBUG = "DEBUG" 

292 INFO = "INFO" 

293 WARNING = "WARNING" 

294 ERROR = "ERROR" 

295 CRITICAL = "CRITICAL" 

296 

297 

298class LogFormat(StrEnum): 

299 """ログフォーマットの定義""" 

300 

301 CONSOLE = "console" 

302 JSON = "json" 

303 

304 

305# ============================================================================= 

306# API設定モデル 

307# ============================================================================= 

308 

309 

310class APIConfig(BaseModel): 

311 """API関連の設定""" 

312 

313 base_url: str = Field( 

314 default="https://jsonplaceholder.typicode.com", 

315 description="APIのベースURL", 

316 ) 

317 timeout: float = Field( 

318 default=30.0, 

319 ge=1.0, 

320 le=300.0, 

321 description="リクエストタイムアウト(秒)", 

322 ) 

323 retry_count: int = Field(default=3, ge=0, le=10, description="リトライ回数") 

324 retry_delay: float = Field(default=1.0, ge=0.1, le=60.0, description="リトライ間隔(秒)") 

325 max_connections: int = Field(default=10, ge=1, le=100, description="最大同時接続数") 

326 user_agent: str = Field( 

327 default="api-test-devops-portfolio/0.1.0", description="User-Agentヘッダー" 

328 ) 

329 

330 @field_validator("base_url") 

331 @classmethod 

332 def validate_base_url(cls, v: str) -> str: 

333 """ベースURLのバリデーション(SSRF Prevention対応) 

334 

335 Security: 

336 - プライベートIP/ループバックアドレスをブロック 

337 - 許可されたドメインのみ許可(ALLOWED_DOMAINS) 

338 - AWS metadata endpoint (169.254.169.254) をブロック 

339 """ 

340 return _validate_base_url_with_allowed_domains(v, ALLOWED_DOMAINS) 

341 

342 

343# ============================================================================= 

344# ログ設定モデル 

345# ============================================================================= 

346 

347 

348class LogConfig(BaseModel): 

349 """ログ関連の設定""" 

350 

351 level: LogLevel = Field(default=LogLevel.INFO, description="ログレベル") 

352 format: LogFormat = Field(default=LogFormat.JSON, description="ログフォーマット") 

353 file: str | None = Field(default=None, description="ログファイルのパス") 

354 max_size: int = Field( 

355 default=10 * 1024 * 1024, # 10MB 

356 ge=1024, 

357 description="ログファイルの最大サイズ(バイト)", 

358 ) 

359 backup_count: int = Field( 

360 default=5, 

361 ge=1, 

362 le=100, 

363 description="ローテーションするログファイル数", 

364 ) 

365 console_output: bool = Field(default=True, description="コンソール出力の有効化") 

366 

367 @field_validator("file") 

368 @classmethod 

369 def validate_log_file(cls, v: str | None) -> str | None: 

370 """ログファイルパスのバリデーション""" 

371 if v is not None: 

372 log_path = Path(v) 

373 # ディレクトリが存在しない場合は作成 

374 log_path.parent.mkdir(parents=True, exist_ok=True) 

375 return v 

376 

377 

378# ============================================================================= 

379# テスト設定モデル 

380# ============================================================================= 

381 

382 

383class TestConfig(BaseModel): 

384 """テスト関連の設定""" 

385 

386 slow_test_threshold: float = Field( 

387 default=5.0, 

388 ge=0.1, 

389 description="スローテストの判定閾値(秒)", 

390 ) 

391 max_concurrent_requests: int = Field( 

392 default=5, 

393 ge=1, 

394 le=50, 

395 description="並行テストでの最大リクエスト数", 

396 ) 

397 external_api_enabled: bool = Field(default=True, description="外部APIテストの有効化") 

398 performance_test_enabled: bool = Field( 

399 default=False, 

400 description="パフォーマンステストの有効化", 

401 ) 

402 security_test_enabled: bool = Field(default=False, description="セキュリティテストの有効化") 

403 test_data_cleanup: bool = Field(default=True, description="テスト後のデータクリーンアップ") 

404 

405 

406# ============================================================================= 

407# セキュリティ設定モデル 

408# ============================================================================= 

409 

410 

411class SecurityConfig(BaseModel): 

412 """セキュリティ関連の設定""" 

413 

414 api_key: SecretStr | None = Field(default=None, description="API認証キー") 

415 jwt_secret: SecretStr | None = Field( 

416 default=None, 

417 description="JWT署名用シークレット", 

418 ) 

419 allowed_hosts: list[str] = Field( 

420 default=["localhost", "127.0.0.1"], 

421 description="許可するホスト一覧", 

422 ) 

423 rate_limit_requests: int = Field(default=100, ge=1, description="レート制限:リクエスト数") 

424 rate_limit_window: int = Field( 

425 default=3600, # 1時間 

426 ge=60, 

427 description="レート制限:時間窓(秒)", 

428 ) 

429 enable_cors: bool = Field(default=False, description="CORS有効化") 

430 

431 

432class SentryConfig(BaseModel): 

433 """Sentry関連の設定""" 

434 

435 dsn: SecretStr = Field(default=SecretStr(""), description="Sentry DSN") 

436 enabled: bool = Field(default=False, description="Sentry有効化") 

437 environment: str | None = Field( 

438 default=None, 

439 description="環境名(未設定時はsettings.environmentを使用)", 

440 ) 

441 traces_sample_rate: float = Field( 

442 default=0.1, 

443 ge=0.0, 

444 le=1.0, 

445 description="トレースサンプリングレート", 

446 ) 

447 profiles_sample_rate: float = Field( 

448 default=0.1, 

449 ge=0.0, 

450 le=1.0, 

451 description="プロファイリングサンプリングレート", 

452 ) 

453 send_default_pii: bool = Field(default=False, description="PII送信の有効化") 

454 

455 

456# ============================================================================= 

457# メイン設定クラス 

458# ============================================================================= 

459 

460 

461# 本番相当環境セット(production + staging) 

462# validate_production_secrets / validate_production_https / is_production_like で共有 

463_PRODUCTION_LIKE_ENVIRONMENTS: frozenset[Environment] = frozenset( 

464 { 

465 Environment.PRODUCTION, 

466 Environment.STAGING, 

467 } 

468) 

469 

470 

471class Settings(BaseSettings): 

472 """アプリケーション設定の統合管理""" 

473 

474 # テスト用フォーマット問題:shorter line 

475 model_config = SettingsConfigDict( 

476 # 環境変数の設定 

477 env_file=".env", 

478 env_file_encoding="utf-8", 

479 env_nested_delimiter="__", # API__BASE_URL のようなネスト記法 

480 case_sensitive=False, 

481 extra="ignore", # 不明な設定値は無視 

482 ) 

483 

484 # 基本設定 

485 environment: Environment = Field(default=Environment.DEVELOPMENT, description="実行環境") 

486 debug: bool = Field(default=False, description="デバッグモードの有効化") 

487 project_name: str = Field(default="API Test Portfolio", description="プロジェクト名") 

488 version: str = Field(default="0.1.0", description="アプリケーションバージョン") 

489 

490 # 各種設定 

491 api: APIConfig = Field(default_factory=APIConfig) 

492 log: LogConfig = Field(default_factory=LogConfig) 

493 test: TestConfig = Field(default_factory=TestConfig) 

494 security: SecurityConfig = Field(default_factory=SecurityConfig) 

495 sentry: SentryConfig = Field(default_factory=SentryConfig) 

496 

497 @field_validator("environment", mode="before") 

498 @classmethod 

499 def validate_environment(cls, v: str | Environment) -> Environment: 

500 """環境設定のバリデーション 

501 

502 Note: 

503 StrEnumはstrを継承するため isinstance(v, str) はEnumインスタンスにもTrueを返す。 

504 Enumインスタンスを先に判定し、strは正規化(strip + lower)してから 

505 Environmentコンストラクタで変換する。これにより末尾スペースやタイポを 

506 早期検出し、わかりやすいエラーメッセージを提供できる。 

507 

508 Raises: 

509 ValueError: 無効な環境名(タイポ・末尾スペース含む)またはstr/Environment以外の型 

510 """ 

511 # StrEnumはstrを継承するため、Enumインスタンスを先に判定して早期リターン 

512 if isinstance(v, Environment): 

513 return v 

514 if isinstance(v, str): 

515 normalized = v.strip().lower() 

516 try: 

517 return Environment(normalized) 

518 except ValueError: 

519 valid = [e.value for e in Environment] 

520 # from None: PydanticがValidationErrorでラップするため 

521 # 元のValueErrorチェーンを隠してエラーメッセージをクリーンに保つ 

522 raise ValueError(f"environment の値が無効です: {v!r}。有効な値: {valid}") from None 

523 # NOTE: mode="before" のためPydantic型強制前に実行され、int/None等も到達可能。 

524 # Pydanticがmode="after"なら非str型は型強制段階で排除されるが、 

525 # mode="before"では生の値を受け取るため、この分岐は防御的コードとして機能する。 

526 raise ValueError( 

527 f"environment には str または Environment を指定してください。" 

528 f"受け取った型: {type(v).__name__!r}, 値: {v!r}" 

529 ) 

530 

531 @model_validator(mode="after") 

532 def validate_production_secrets(self) -> Self: 

533 """本番・ステージング環境でのシークレット存在チェック 

534 

535 Security: 

536 本番環境(ENVIRONMENT=production)およびステージング(ENVIRONMENT=staging)では、 

537 api_keyまたはjwt_secretの少なくとも一方が必須。 

538 未設定の場合はValueErrorを発生させ、サイレント失敗を防止。 

539 ステージングは本番と同等のインフラ・データを扱うため同一ポリシーを適用。 

540 

541 Note: 

542 開発/テスト環境ではシークレットなしで動作可能。 

543 

544 """ 

545 if self.environment in _PRODUCTION_LIKE_ENVIRONMENTS: 

546 if self.security.api_key is None and self.security.jwt_secret is None: 

547 raise ValueError( 

548 f"{self.environment.value.capitalize()} environment requires at least one of: " 

549 "SECURITY__API_KEY or SECURITY__JWT_SECRET", 

550 ) 

551 return self 

552 

553 @model_validator(mode="after") 

554 def validate_production_https(self) -> Self: 

555 """本番・ステージング環境でのHTTPS強制 

556 

557 Security: 

558 本番(ENVIRONMENT=production)およびステージング(ENVIRONMENT=staging)では、 

559 HTTPS接続を強制し平文HTTP通信を禁止。 

560 OWASP A02: Cryptographic Failures対策。 

561 ステージングは本番と同等のインフラ・データを扱うため同一ポリシーを適用。 

562 

563 Note: 

564 開発/テスト環境ではHTTP接続を許可(ローカル開発用)。 

565 """ 

566 if self.environment in _PRODUCTION_LIKE_ENVIRONMENTS: 

567 if not self.api.base_url.startswith("https://"): 

568 raise ValueError( 

569 f"{self.environment.value.capitalize()} environment requires HTTPS. " 

570 "Set API__BASE_URL to an https:// URL.", 

571 ) 

572 return self 

573 

574 def is_development(self) -> bool: 

575 """開発環境判定""" 

576 return self.environment == Environment.DEVELOPMENT 

577 

578 def is_testing(self) -> bool: 

579 """テスト環境判定""" 

580 return self.environment == Environment.TESTING 

581 

582 def is_production(self) -> bool: 

583 """本番環境判定""" 

584 return self.environment == Environment.PRODUCTION 

585 

586 def is_staging(self) -> bool: 

587 """ステージング環境判定""" 

588 return self.environment == Environment.STAGING 

589 

590 def is_production_like(self) -> bool: 

591 """本番相当環境判定(production + staging)""" 

592 return self.environment in _PRODUCTION_LIKE_ENVIRONMENTS 

593 

594 def get_log_level(self) -> int: 

595 """ログレベルの取得(logging/structlog共通) 

596 

597 structlogはloggingモジュールのログレベル定数を使用するため、 

598 標準loggingの定数をそのまま返す。 

599 

600 Returns: 

601 int: logging.DEBUG (10), INFO (20), WARNING (30), ERROR (40), CRITICAL (50) 

602 

603 Example: 

604 >>> import structlog 

605 >>> structlog.configure( 

606 ... wrapper_class=structlog.make_filtering_bound_logger(settings.get_log_level()) 

607 ... ) 

608 

609 """ 

610 # LogLevel Enum → logging定数への明示的マッピング(型安全) 

611 level_map: dict[LogLevel, int] = { 

612 LogLevel.DEBUG: logging.DEBUG, 

613 LogLevel.INFO: logging.INFO, 

614 LogLevel.WARNING: logging.WARNING, 

615 LogLevel.ERROR: logging.ERROR, 

616 LogLevel.CRITICAL: logging.CRITICAL, 

617 } 

618 return level_map[self.log.level] 

619 

620 def to_dict(self, exclude_secrets: bool = True) -> dict[str, Any]: 

621 """設定を辞書形式で出力""" 

622 data = self.model_dump() 

623 

624 if exclude_secrets: 

625 # シークレット値をマスク 

626 if "security" in data: 626 ↛ 634line 626 didn't jump to line 634 because the condition on line 626 was always true

627 security = data["security"] 

628 if security.get("api_key"): 

629 security["api_key"] = "***MASKED***" 

630 if security.get("jwt_secret"): 

631 security["jwt_secret"] = "***MASKED***" # noqa: S105 

632 

633 # Sentry DSNをマスク(空文字列も含めてマスク) 

634 if "sentry" in data: 634 ↛ 639line 634 didn't jump to line 639 because the condition on line 634 was always true

635 sentry = data["sentry"] 

636 if "dsn" in sentry: 636 ↛ 639line 636 didn't jump to line 639 because the condition on line 636 was always true

637 sentry["dsn"] = "***MASKED***" 

638 

639 return data 

640 

641 

642# ============================================================================= 

643# グローバル設定インスタンス 

644# ============================================================================= 

645 

646# シングルトンパターンで設定を管理 

647_settings: Settings | None = None 

648 

649 

650def get_settings() -> Settings: 

651 """設定インスタンスの取得(シングルトン)""" 

652 global _settings 

653 if _settings is None: 

654 _settings = Settings() 

655 return _settings 

656 

657 

658def reload_settings() -> Settings: 

659 """設定の再読み込み(主にテスト用)""" 

660 global _settings 

661 _settings = Settings() 

662 return _settings 

663 

664 

665# 便利なエイリアス 

666# モジュールインポート時に ValidationError が発生すると ImportError として連鎖し 

667# 根本原因(環境変数のタイポ等)が隠蔽される。logging で根本原因を明示する。 

668try: 

669 settings = get_settings() 

670except Exception as e: # noqa: BLE001 # ValidationError含む全設定エラーを捕捉(SystemExit等のBaseException除外済み) 

671 _logger.critical( 

672 "設定の初期化に失敗しました (type=%s)。環境変数を確認してください: %s", 

673 type(e).__name__, 

674 e, 

675 ) 

676 raise