You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TimeoutConfig:
|
|
connect: float = 1.0
|
|
read: float = 5.0
|
|
write: float = 5.0
|
|
pool: float = 1.0
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: dict[str, Any] | None) -> "TimeoutConfig":
|
|
value = value or {}
|
|
return cls(
|
|
connect=float(value.get("connect", cls.connect)),
|
|
read=float(value.get("read", cls.read)),
|
|
write=float(value.get("write", cls.write)),
|
|
pool=float(value.get("pool", cls.pool)),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RetryConfig:
|
|
attempts: int = 1
|
|
backoff: float = 0.2
|
|
statuses: tuple[int, ...] = field(default_factory=lambda: (502, 503, 504))
|
|
methods: tuple[str, ...] = field(default_factory=lambda: ("GET", "HEAD", "OPTIONS"))
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: dict[str, Any] | None) -> "RetryConfig":
|
|
value = value or {}
|
|
return cls(
|
|
attempts=max(int(value.get("attempts", cls.attempts)), 1),
|
|
backoff=float(value.get("backoff", cls.backoff)),
|
|
statuses=tuple(int(item) for item in value.get("statuses", cls().statuses)),
|
|
methods=tuple(str(item).upper() for item in value.get("methods", cls().methods)),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CircuitBreakerConfig:
|
|
enabled: bool = False
|
|
fail_max: int = 5
|
|
reset_timeout: int = 30
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: dict[str, Any] | None) -> "CircuitBreakerConfig":
|
|
value = value or {}
|
|
return cls(
|
|
enabled=bool(value.get("enabled", cls.enabled)),
|
|
fail_max=max(int(value.get("fail_max", cls.fail_max)), 1),
|
|
reset_timeout=max(int(value.get("reset_timeout", cls.reset_timeout)), 1),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServiceConfig:
|
|
name: str
|
|
base_url: str
|
|
token: str | None = None
|
|
timeout: TimeoutConfig = field(default_factory=TimeoutConfig)
|
|
retry: RetryConfig = field(default_factory=RetryConfig)
|
|
circuit_breaker: CircuitBreakerConfig = field(default_factory=CircuitBreakerConfig)
|
|
|
|
@classmethod
|
|
def from_mapping(cls, name: str, value: dict[str, Any]) -> "ServiceConfig":
|
|
return cls(
|
|
name=name,
|
|
base_url=str(value["base_url"]).rstrip("/"),
|
|
token=value.get("token"),
|
|
timeout=TimeoutConfig.from_mapping(value.get("timeout")),
|
|
retry=RetryConfig.from_mapping(value.get("retry")),
|
|
circuit_breaker=CircuitBreakerConfig.from_mapping(
|
|
value.get("circuit_breaker")
|
|
),
|
|
)
|