Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 from __future__ import annotations 2 3 import time 4 from typing import Any, Literal, Optional 5 from uuid import UUID 6 7 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator 8 9 10 def now_ts_ms() -> int: 11 return int(time.time() * 1000) 12 13 14 class ErrorInfo(BaseModel): 15 model_config = ConfigDict(extra="forbid") 16 17 code: str = Field(min_length=1) 18 message: str = Field(min_length=1) 19 details: Optional[dict[str, Any]] = None 20 21 22 class TraceEnvelope(BaseModel): 23 """ 24 Base envelope required on all inter-service payloads. 25 26 Notes: 27 - `correlation_id` is optional; if not provided, it should be treated as `trace_id`. 28 """ 29 30 model_config = ConfigDict(extra="forbid") 31 32 trace_id: UUID 33 request_id: UUID 34 correlation_id: Optional[UUID] = None 35 session_id: Optional[UUID] = None 36 ts_ms: int = Field(ge=0) 37 38 @field_validator("ts_ms") 39 @classmethod 40 def _validate_ts_ms(cls, v: int) -> int: 41 if v <= 0: 42 raise ValueError("ts_ms must be a positive unix timestamp in milliseconds") 43 return v 44 45 @model_validator(mode="after") 46 def _default_correlation_id(self) -> "TraceEnvelope": 47 if self.correlation_id is None: 48 self.correlation_id = self.trace_id 49 return self 50 51 52 class ServiceHealthRequest(TraceEnvelope): 53 model_config = ConfigDict(extra="forbid") 54 55 kind: Literal["health"] = "health" 56 57 58 class ServiceHealthResponse(TraceEnvelope): 59 model_config = ConfigDict(extra="forbid") 60 61 ok: bool 62 status: Literal["healthy", "unhealthy"] 63 details: dict[str, Any] = Field(default_factory=dict) 64 error: Optional[ErrorInfo] = None