https://pastein.ru/t/aAa
Загрузка данных
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import abc
import argparse
import atexit
import contextlib
import contextvars
import dataclasses
import datetime
import enum
import functools
import hashlib
import json
import logging
import sys
import threading
import time
import traceback
import uuid
from typing import (
Any,
Callable,
Dict,
Final,
FrozenSet,
Generic,
Iterable,
List,
Mapping,
Optional,
Protocol,
Sequence,
Tuple,
Type,
TypeVar,
cast,
final,
runtime_checkable,
)
__version__: Final[str] = "1.0.0"
__all__: Final[Tuple[str, ...]] = (
"Greeting",
"RenderRequest",
"RenderResult",
"HelloWorldApplication",
"ServiceContainer",
"EventBus",
"build_application",
"main",
)
# =============================================================================
# РАЗДЕЛ 1. Константы предметной области
# =============================================================================
DEFAULT_SALUTATION: Final[str] = "Hello"
DEFAULT_TARGET: Final[str] = "World"
DEFAULT_PUNCTUATION: Final[str] = "!"
MAX_SALUTATION_LENGTH: Final[int] = 128
MAX_TARGET_LENGTH: Final[int] = 256
MIN_PUNCTUATION_LENGTH: Final[int] = 1
LOGGER_NAME: Final[str] = "hello_world"
LOGGER: Final[logging.Logger] = logging.getLogger(LOGGER_NAME)
_METRIC_LOCK: Final[threading.Lock] = threading.Lock()
# =============================================================================
# РАЗДЕЛ 2. Иерархия исключений предметной области
# =============================================================================
class HelloWorldError(Exception):
"""Базовый класс всех ошибок предметной области.
Каждая ошибка снабжается уникальным идентификатором, что позволяет
коррелировать сообщения пользователя с записями журнала.
"""
def __init__(
self,
message: str,
*,
details: Optional[Mapping[str, Any]] = None,
) -> None:
super().__init__(message)
self.message: str = message
self.details: Mapping[str, Any] = dict(details or {})
self.error_id: str = uuid.uuid4().hex
self.occurred_at: datetime.datetime = datetime.datetime.now(
datetime.timezone.utc
)
def to_dict(self) -> Dict[str, Any]:
"""Сериализует ошибку в словарь для структурного журнала."""
return {
"error_id": self.error_id,
"type": type(self).__name__,
"message": self.message,
"details": dict(self.details),
"occurred_at": self.occurred_at.isoformat(),
}
def __str__(self) -> str:
base = f"[{self.error_id[:8]}] {self.message}"
if self.details:
rendered = ", ".join(f"{key}={value!r}" for key, value in self.details.items())
return f"{base} ({rendered})"
return base
class ConfigurationError(HelloWorldError):
"""Ошибка конфигурации приложения, обнаруженная при сборке контейнера."""
class ValidationError(HelloWorldError):
"""Нарушение инвариантов value object ``Greeting``."""
class CompositeValidationError(ValidationError):
"""Агрегированная ошибка, объединяющая несколько нарушений валидации."""
def __init__(self, errors: Sequence[ValidationError]) -> None:
flattened: List[str] = [error.message for error in errors]
super().__init__(
"Обнаружено нарушений валидации: " + "; ".join(flattened),
details={"count": len(errors)},
)
self.errors: Tuple[ValidationError, ...] = tuple(errors)
class ProviderNotFoundError(ConfigurationError):
"""Запрошенный провайдер отсутствует в реестре."""
class DependencyNotFoundError(ConfigurationError):
"""Зависимость не зарегистрирована в контейнере инверсии зависимостей."""
class CircularDependencyError(ConfigurationError):
"""Обнаружена циклическая зависимость при резолюции сервисов."""
class RenderingError(HelloWorldError):
"""Ошибка на этапе формирования итогового сообщения."""
class ChannelClosedError(HelloWorldError):
"""Попытка записи в закрытый канал вывода."""
class RetryBudgetExhaustedError(HelloWorldError):
"""Все повторные попытки выполнения операции исчерпаны."""
class InvalidStateTransitionError(HelloWorldError):
"""Попытка нелегального перехода в жизненном цикле приложения."""
# =============================================================================
# РАЗДЕЛ 3. Перечисления (enums)
# =============================================================================
class OutputChannelType(enum.Enum):
"""Тип канала вывода сообщения."""
STDOUT = "stdout"
STDERR = "stderr"
BUFFER = "buffer"
class MessageFormat(enum.Enum):
"""Стратегия форматирования текста сообщения."""
PLAIN = "plain"
UPPER = "upper"
LOWER = "lower"
TITLE = "title"
SHOUTING = "shouting"
class Verbosity(enum.IntEnum):
"""Уровень детализации сопутствующего вывода."""
QUIET = 0
NORMAL = 1
VERBOSE = 2
DEBUG = 3
class ServiceLifetime(enum.Enum):
"""Время жизни сервиса в контейнере зависимостей."""
SINGLETON = "singleton"
TRANSIENT = "transient"
class LifecycleState(enum.Enum):
"""Этапы жизненного цикла приложения."""
CREATED = "created"
INITIALIZED = "initialized"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
def allows_transition_to(self, other: "LifecycleState") -> bool:
"""Проверяет, является ли переход ``self -> other`` легальным."""
allowed: Mapping[LifecycleState, FrozenSet[LifecycleState]] = {
LifecycleState.CREATED: {LifecycleState.INITIALIZED, LifecycleState.FAILED},
LifecycleState.INITIALIZED: {LifecycleState.RUNNING, LifecycleState.FAILED},
LifecycleState.RUNNING: {LifecycleState.COMPLETED, LifecycleState.FAILED},
LifecycleState.COMPLETED: frozenset(),
LifecycleState.FAILED: frozenset(),
}
return other in allowed[self]
# =============================================================================
# РАЗДЕЛ 4. Неизменяемые value objects
# =============================================================================
@final
@dataclasses.dataclass(frozen=True, repr=True)
class Greeting:
"""Неизменяемое представление приветствия.
Атрибуты:
salutation: приветственная фраза (например, «Hello»).
target: адресат приветствия (например, «World»).
punctuation: завершающая пунктуация.
"""
salutation: str = DEFAULT_SALUTATION
target: str = DEFAULT_TARGET
punctuation: str = DEFAULT_PUNCTUATION
def __post_init__(self) -> None:
"""Проверяет инварианты value object в момент создания."""
if not isinstance(self.salutation, str) or not self.salutation.strip():
raise ValidationError("Параметр salutation должен быть непустой строкой")
if not isinstance(self.target, str) or not self.target.strip():
raise ValidationError("Параметр target должен быть непустой строкой")
if len(self.salutation) > MAX_SALUTATION_LENGTH:
raise ValidationError(
"Длина salutation превышает допустимую",
details={"limit": MAX_SALUTATION_LENGTH},
)
if len(self.target) > MAX_TARGET_LENGTH:
raise ValidationError(
"Длина target превышает допустимую",
details={"limit": MAX_TARGET_LENGTH},
)
if len(self.punctuation) < MIN_PUNCTUATION_LENGTH:
raise ValidationError("Пунктуация не может быть пустой")
def compose(self) -> str:
"""Собирает каноническую строку приветствия."""
return f"{self.salutation}, {self.target}{self.punctuation}"
def to_dict(self) -> Dict[str, str]:
"""Возвращает словарное представление для сериализации."""
return {
"salutation": self.salutation,
"target": self.target,
"punctuation": self.punctuation,
}
@final
@dataclasses.dataclass(frozen=True)
class RenderRequest:
"""Неизменяемый снимок параметров одного рендеринга."""
greeting: Greeting
message_format: MessageFormat = MessageFormat.PLAIN
verbosity: Verbosity = Verbosity.NORMAL
correlation_id: str = dataclasses.field(default_factory=lambda: uuid.uuid4().hex)
@final
@dataclasses.dataclass(frozen=True)
class RenderResult:
"""Иммутабельный результат рендеринга и доставки сообщения."""
payload: str
byte_length: int
checksum: str
duration_ms: float
correlation_id: str
def to_dict(self) -> Dict[str, Any]:
"""Подготавливает результат к сериализации в JSON."""
return {
"payload": self.payload,
"byte_length": self.byte_length,
"checksum": self.checksum,
"duration_ms": round(self.duration_ms, 4),
"correlation_id": self.correlation_id,
}
# =============================================================================
# РАЗДЕЛ 5. Протоколы (структурные интерфейсы)
# =============================================================================
@runtime_checkable
class MessageFormatter(Protocol):
"""Стратегия преобразования текста сообщения."""
def format(self, text: str) -> str:
"""Преобразует текст сообщения согласно правилам стратегии."""
...
@runtime_checkable
class OutputChannel(Protocol):
"""Абстракция канала доставки сообщения."""
@property
def is_open(self) -> bool:
"""Признак того, что канал открыт для записи."""
...
def emit(self, payload: str) -> None:
"""Доставляет полезную нагрузку в канал."""
...
def flush(self) -> None:
"""Сбрасывает буферы канала."""
...
@runtime_checkable
class Clock(Protocol):
"""Абстракция источника времени — необходима для детерминированных тестов."""
def now(self) -> datetime.datetime:
"""Возвращает текущие дата и время с часовым поясом."""
...
def monotonic(self) -> float:
"""Возвращает монотонные часы в секундах."""
...
@runtime_checkable
class TelemetrySink(Protocol):
"""Абстракция приёмника телеметрических метрик."""
def record(self, metric: str, value: float, tags: Mapping[str, str]) -> None:
"""Регистрирует значение метрики с произвольными тегами."""
...
# =============================================================================
# РАЗДЕЛ 6. Абстрактные базовые классы предметной области
# =============================================================================
class GreetingProvider(abc.ABC):
"""Поставщик приветствий: единственная точка получения доменных данных."""
REGISTRY_KEY: str = ""
@abc.abstractmethod
def provide(self) -> Greeting:
"""Возвращает валидное приветствие."""
raise NotImplementedError
def describe(self) -> str:
"""Человекочитаемое описание провайдера (может быть переопределено)."""
return f"{type(self).__name__}(key={self.REGISTRY_KEY!r})"
class RenderingStrategy(abc.ABC):
"""Абстрактная стратегия превращения запроса в строку."""
@abc.abstractmethod
def render(self, request: RenderRequest) -> str:
"""Выполняет рендеринг сообщения по переданному запросу."""
raise NotImplementedError
class Validator(abc.ABC):
"""Абстрактный валидатор приветствий."""
@abc.abstractmethod
def validate(self, greeting: Greeting) -> None:
"""Поднимает ``ValidationError``, если инварианты нарушены."""
raise NotImplementedError
# =============================================================================
# РАЗДЕЛ 7. Реестр провайдеров на основе метакласса
# =============================================================================
class ProviderRegistryMeta(abc.ABCMeta):
"""Метакласс, автоматически регистрирующий конкретных провайдеров.
Любой класс-наследник ``GreetingProvider``, реализовавший все абстрактные
методы и объявивший непустой ``REGISTRY_KEY``, попадает в реестр без
какого-либо дополнительного кода со стороны разработчика.
"""
_registry: Dict[str, Type["GreetingProvider"]] = {}
def __init__(
cls,
name: str,
bases: Tuple[type, ...],
namespace: Dict[str, Any],
) -> None:
super().__init__(name, bases, namespace)
abstract: FrozenSet[str] = getattr(cls, "__abstractmethods__", frozenset())
key: str = getattr(cls, "REGISTRY_KEY", "")
if not abstract and key:
ProviderRegistryMeta._registry[key] = cls
LOGGER.debug("Провайдер %s зарегистрирован под ключом %r", name, key)
@classmethod
def get(cls, key: str) -> Type["GreetingProvider"]:
"""Возвращает класс провайдера по ключу или поднимает ошибку."""
try:
return ProviderRegistryMeta._registry[key]
except KeyError:
raise ProviderNotFoundError(
"Провайдер не найден",
details={"key": key, "available": sorted(ProviderRegistryMeta._registry)},
) from None
@classmethod
def keys(cls) -> Tuple[str, ...]:
"""Возвращает отсортированный перечень доступных ключей."""
return tuple(sorted(ProviderRegistryMeta._registry))
class RegisteredGreetingProvider(GreetingProvider, metaclass=ProviderRegistryMeta):
"""Базовый класс с метаклассом-реестром; сам по себе нерегистрируется."""
@final
class DefaultGreetingProvider(RegisteredGreetingProvider):
"""Канонический провайдер: «Hello, World!»."""
REGISTRY_KEY = "default"
def provide(self) -> Greeting:
"""Возвращает классическое приветствие всемирно известного образца."""
return Greeting(
salutation=DEFAULT_SALUTATION,
target=DEFAULT_TARGET,
punctuation=DEFAULT_PUNCTUATION,
)
@final
class FormalGreetingProvider(RegisteredGreetingProvider):
"""Вежливый провайдер с увеличенной пунктуацией."""
REGISTRY_KEY = "formal"
def provide(self) -> Greeting:
"""Возвращает вежливое приветствие."""
return Greeting(salutation="Good day", target="dear World", punctuation=".")
@final
class EnthusiasticGreetingProvider(RegisteredGreetingProvider):
"""Эмоционально окрашенный провайдер."""
REGISTRY_KEY = "enthusiastic"
def provide(self) -> Greeting:
"""Возвращает приветствие с восторженной пунктуацией."""
return Greeting(salutation="Hello", target="Wonderful World", punctuation="!!!")
# =============================================================================
# РАЗДЕЛ 8. Конкретные реализации стратегий форматирования
# =============================================================================
class PlainFormatter:
"""Стратегия, не изменяющая текст (стратегия по умолчанию)."""
def format(self, text: str) -> str:
"""Возвращает текст без изменений."""
return text
class UpperFormatter:
"""Стратегия, переводящая текст в верхний регистр."""
def format(self, text: str) -> str:
"""Возвращает текст в верхнем регистре."""
return text.upper()
class LowerFormatter:
"""Стратегия, переводящая текст в нижний регистр."""
def format(self, text: str) -> str:
"""Возвращает текст в нижнем регистре."""
return text.lower()
class TitleFormatter:
"""Стратегия, приводящая текст к «заглавному» регистру."""
def format(self, text: str) -> str:
"""Возвращает текст в регистре Title Case."""
return text.title()
class ShoutingFormatter:
"""Стратегия, имитирующая крик (верхний регистр + многоточие)."""
def format(self, text: str) -> str:
"""Возвращает текст, написанный капсом через дефис."""
separator: Final[str] = "-"
return separator.join(text.upper())
FORMATTER_REGISTRY: Final[Mapping[MessageFormat, Callable[[], MessageFormatter]]] = {
MessageFormat.PLAIN: PlainFormatter,
MessageFormat.UPPER: UpperFormatter,
MessageFormat.LOWER: LowerFormatter,
MessageFormat.TITLE: TitleFormatter,
MessageFormat.SHOUTING: ShoutingFormatter,
}
def create_formatter(message_format: MessageFormat) -> MessageFormatter:
"""Фабричная функция, возвращающая стратегию по значению перечисления."""
try:
factory = FORMATTER_REGISTRY[message_format]
except KeyError:
raise ConfigurationError(
"Неизвестный формат сообщения",
details={"format": message_format},
) from None
formatter = factory()
LOGGER.debug("Создана стратегия форматирования: %s", type(formatter).__name__)
return formatter
# =============================================================================
# РАЗДЕЛ 9. Реализации часов
# =============================================================================
@final
class SystemClock:
"""Реальная реализация часов на основе модуля ``datetime`` и ``time``."""
def now(self) -> datetime.datetime:
"""Возвращает текущие UTC-дата и время."""
return datetime.datetime.now(datetime.timezone.utc)
def monotonic(self) -> float:
"""Возвращает значение монотонных часов."""
return time.perf_counter()
@final
class FrozenClock:
"""Детерминированные часы для воспроизводимых тестов."""
def __init__(self, instant: datetime.datetime) -> None:
self._instant: datetime.datetime = instant
self._ticks: float = 0.0
def now(self) -> datetime.datetime:
"""Возвращает зафиксированное значение времени."""
return self._instant
def monotonic(self) -> float:
"""Возвращает искусственно увеличивающееся значение."""
self._ticks += 1.0
return self._ticks
# =============================================================================
# РАЗДЕЛ 10. Каналы вывода
# =============================================================================
class AbstractOutputChannel:
"""Общая логика каналов вывода: состояние, счётчики, блокировки."""
def __init__(self, name: str) -> None:
self._name: str = name
self._open: bool = True
self._emitted: int = 0
self._lock: threading.RLock = threading.RLock()
@property
def name(self) -> str:
"""Наименование канала (используется в журнале и метриках)."""
return self._name
@property
def is_open(self) -> bool:
"""Признак открытого канала."""
with self._lock:
return self._open
@property
def emitted_count(self) -> int:
"""Число успешно доставленных сообщений."""
with self._lock:
return self._emitted
def close(self) -> None:
"""Закрывает канал, запрещая дальнейшую запись."""
with self._lock:
self._open = False
LOGGER.debug("Канал %s закрыт", self._name)
def _ensure_open(self) -> None:
"""Поднимает ``ChannelClosedError``, если канал закрыт."""
if not self.is_open:
raise ChannelClosedError(
"Канал вывода закрыт",
details={"channel": self._name},
)
def _account_emitted(self) -> None:
"""Инкрементирует счётчик доставленных сообщений потокобезопасно."""
with self._lock:
self._emitted += 1
def flush(self) -> None:
"""Переопределяется производными классами при наличии буферов."""
return None
@final
class StandardOutputChannel(AbstractOutputChannel):
"""Канал, записывающий сообщения в ``sys.stdout``."""
def __init__(self) -> None:
super().__init__("stdout")
def emit(self, payload: str) -> None:
"""Записывает строку в стандартный поток вывода и сбрасывает буфер."""
self._ensure_open()
sys.stdout.write(payload + "\n")
sys.stdout.flush()
self._account_emitted()
LOGGER.debug("Доставлено в stdout: %r", payload)
def flush(self) -> None:
"""Сбрасывает буфер стандартного потока вывода."""
sys.stdout.flush()
@final
class StandardErrorChannel(AbstractOutputChannel):
"""Канал, записывающий сообщения в ``sys.stderr``."""
def __init__(self) -> None:
super().__init__("stderr")
def emit(self, payload: str) -> None:
"""Записывает строку в поток ошибок."""
self._ensure_open()
sys.stderr.write(payload + "\n")
sys.stderr.flush()
self._account_emitted()
@final
class BufferChannel(AbstractOutputChannel):
"""Канал, накапливающий сообщения в памяти (для тестов и батчинга)."""
def __init__(self) -> None:
super().__init__("buffer")
self._buffer: List[str] = []
def emit(self, payload: str) -> None:
"""Накапливает полезную нагрузку во внутреннем буфере."""
self._ensure_open()
with self._lock:
self._buffer.append(payload)
self._account_emitted()
def drain(self) -> Tuple[str, ...]:
"""Извлекает и очищает содержимое буфера."""
with self._lock:
drained = tuple(self._buffer)
self._buffer.clear()
return drained
CHANNEL_FACTORIES: Final[Mapping[OutputChannelType, Callable[[], AbstractOutputChannel]]] = {
OutputChannelType.STDOUT: StandardOutputChannel,
OutputChannelType.STDERR: StandardErrorChannel,
OutputChannelType.BUFFER: BufferChannel,
}
def create_channel(channel_type: OutputChannelType) -> AbstractOutputChannel:
"""Фабричная функция создания канала вывода."""
try:
factory = CHANNEL_FACTORIES[channel_type]
except KeyError:
raise ConfigurationError(
"Неизвестный тип канала вывода", details={"channel": channel_type}
) from None
return factory()
# =============================================================================
# РАЗДЕЛ 11. Цепочка ответственности: валидаторы
# =============================================================================
class NonEmptyValidator(Validator):
"""Проверяет, что salutation и target не состоят только из пробелов."""
def validate(self, greeting: Greeting) -> None:
"""Поднимает ошибку при пустых составных частях."""
for attribute in ("salutation", "target"):
value = getattr(greeting, attribute)
if not value.strip():
raise ValidationError(
"Поле обязательно к заполнению", details={"field": attribute}
)
class LengthLimitValidator(Validator):
"""Проверяет ограничения на длину составных частей."""
def validate(self, greeting: Greeting) -> None:
"""Поднимает ошибку при превышении лимитов длины."""
limits = (("salutation", MAX_SALUTATION_LENGTH), ("target", MAX_TARGET_LENGTH))
for attribute, limit in limits:
if len(getattr(greeting, attribute)) > limit:
raise ValidationError(
"Превышен допустимый размер поля",
details={"field": attribute, "limit": limit},
)
class AsciiOnlyValidator(Validator):
"""Пример валидатора, который по умолчанию отключён (пустой контракт)."""
def validate(self, greeting: Greeting) -> None:
"""Явно ничего не делает: нелокализованный ввод разрешён."""
return None
class ValidationChain:
"""Цепочка ответственности, агрегирующая всех валидаторов.
Все ошибки собираются и передаются пользователю одной пачкой — так удобнее,
чем чинить по одной за каждый прогон.
"""
def __init__(self, validators: Sequence[Validator]) -> None:
self._validators: Tuple[Validator, ...] = tuple(validators)
@property
def size(self) -> int:
"""Число зарегистрированных валидаторов."""
return len(self._validators)
def validate(self, greeting: Greeting) -> None:
"""Выполняет все проверки, агрегируя ошибки в ``CompositeValidationError``."""
errors: List[ValidationError] = []
for validator in self._validators:
try:
validator.validate(greeting)
except ValidationError as exc:
LOGGER.debug(
"Валидатор %s сообщил об ошибке: %s",
type(validator).__name__,
exc.message,
)
errors.append(exc)
if errors:
raise CompositeValidationError(errors)
LOGGER.debug("Валидация приветствия завершена успешно (%d проверок)", self.size)
def build_default_validation_chain() -> ValidationChain:
"""Собирает каноническую цепочку валидаторов."""
return ValidationChain((NonEmptyValidator(), LengthLimitValidator(), AsciiOnlyValidator()))
# =============================================================================
# РАЗДЕЛ 12. Стратегии рендеринга
# =============================================================================
class CompositeRenderingStrategy(RenderingStrategy):
"""Композитная стратегия: сборка → форматирование → контроль качества."""
def __init__(self, formatter: MessageFormatter) -> None:
self._formatter: MessageFormatter = formatter
def render(self, request: RenderRequest) -> str:
"""Выполняет поэтапный рендеринг сообщения."""
try:
raw = request.greeting.compose()
except Exception as exc:
raise RenderingError(
"Не удалось собрать исходную строку приветствия",
details={"reason": str(exc)},
) from exc
formatted = self._formatter.format(raw)
if not formatted:
raise RenderingError("Результат форматирования пуст")
LOGGER.debug(
"Рендеринг завершён: исходный=%r, итоговый=%r", raw, formatted
)
return formatted
# =============================================================================
# РАЗДЕЛ 13. Телеметрия
# =============================================================================
@final
class InMemoryTelemetrySink:
"""Потокобезопасное хранилище метрик в памяти."""
def __init__(self) -> None:
self._storage: Dict[Tuple[str, Tuple[Tuple[str, str], ...]], List[float]] = {}
self._lock: threading.RLock = threading.RLock()
def record(self, metric: str, value: float, tags: Mapping[str, str]) -> None:
"""Регистрирует значение метрики с нормализованными тегами."""
key = (metric, tuple(sorted(tags.items())))
with self._lock:
self._storage.setdefault(key, []).append(value)
def snapshot(self) -> Dict[str, Any]:
"""Возвращает снимок агрегированных метрик."""
with self._lock:
result: Dict[str, Any] = {}
for (metric, tags), values in sorted(self._storage.items()):
tag_map = dict(tags)
result[f"{metric}|{tag_map}"] = {
"count": len(values),
"sum": round(sum(values), 6),
"min": round(min(values), 6),
"max": round(max(values), 6),
"avg": round(sum(values) / len(values), 6),
}
return result
# =============================================================================
# РАЗДЕЛ 14. Шина событий (наблюдатель)
# =============================================================================
@final
@dataclasses.dataclass(frozen=True)
class Event:
"""Неизменяемое событие предметной области."""
name: str
payload: Mapping[str, Any] = dataclasses.field(default_factory=dict)
occurred_at: datetime.datetime = dataclasses.field(
default_factory=lambda: datetime.datetime.now(datetime.timezone.utc)
)
event_id: str = dataclasses.field(default_factory=lambda: uuid.uuid4().hex)
EventHandler = Callable[[Event], None]
class EventBus:
"""Потокобезопасная шина событий с изоляцией отказов подписчиков."""
def __init__(self, clock: Optional[Clock] = None) -> None:
self._subscribers: Dict[str, List[EventHandler]] = {}
self._lock: threading.RLock = threading.RLock()
self._clock: Clock = clock or SystemClock()
self._published: int = 0
def subscribe(self, event_name: str, handler: EventHandler) -> Callable[[], None]:
"""Подписывает обработчик и возвращает функцию отписки."""
with self._lock:
self._subscribers.setdefault(event_name, []).append(handler)
LOGGER.debug("Подписчик добавлен на событие %r", event_name)
def unsubscribe() -> None:
with self._lock:
handlers = self._subscribers.get(event_name, [])
if handler in handlers:
handlers.remove(handler)
return unsubscribe
def publish(self, event: Event) -> int:
"""Публикует событие, возвращая число успешно обработанных подписок."""
with self._lock:
handlers = tuple(self._subscribers.get(event.name, ()))
self._published += 1
succeeded = 0
for handler in handlers:
try:
handler(event)
succeeded += 1
except Exception:
LOGGER.exception(
"Отказ подписчика %r события %r не прервал публикацию",
getattr(handler, "__qualname__", repr(handler)),
event.name,
)
return succeeded
@property
def published_count(self) -> int:
"""Общее число опубликованных событий."""
with self._lock:
return self._published
def metric_recorder(sink: TelemetrySink, clock: Clock) -> EventHandler:
"""Создаёт обработчик, превращающий события в метрики длительности."""
def handler(event: Event) -> None:
duration = event.payload.get("duration_ms")
if isinstance(duration, (int, float)):
sink.record("event.duration_ms", float(duration), {"event": event.name})
else:
sink.record("event.count", 1.0, {"event": event.name})
return handler
def console_announcer(event: Event) -> None:
"""Демонстрационный подписчик: печатает вспомогательные сообщения."""
LOGGER.info("Событие: %s", event.name)
# =============================================================================
# РАЗДЕЛ 15. Декораторы сопровождения (cross-cutting concerns)
# =============================================================================
_CORRELATION_ID: contextvars.ContextVar[str] = contextvars.ContextVar(
"correlation_id", default=""
)
@contextlib.contextmanager
def trace_context(correlation_id: str) -> Iterable[str]:
"""Устанавливает корреляционный идентификатор в контексте выполнения."""
token = _CORRELATION_ID.set(correlation_id)
try:
yield correlation_id
finally:
_CORRELATION_ID.reset(token)
def traced(
func: Callable[..., Any],
) -> Callable[..., Any]:
"""Декоратор трассировки: журнал входа, выхода и длительности операции."""
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
started = time.perf_counter()
LOGGER.debug("Вход в %s", func.__qualname__)
try:
result = func(*args, **kwargs)
except Exception as exc:
LOGGER.warning(
"Исключение в %s после %.3f мс: %s",
func.__qualname__,
(time.perf_counter() - started) * 1000.0,
exc,
)
raise
LOGGER.debug(
"Выход из %s за %.3f мс",
func.__qualname__,
(time.perf_counter() - started) * 1000.0,
)
return result
return wrapper
def retry(
max_attempts: int = 3,
base_delay: float = 0.001,
retry_on: Tuple[Type[BaseException], ...] = (Exception,),
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""Декоратор повторов с экспоненциальной задержкой.
После исчерпания бюджета повторов поднимается ``RetryBudgetExhaustedError``.
"""
if max_attempts < 1:
raise ConfigurationError("max_attempts должен быть не меньше единицы")
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
last_exception: Optional[BaseException] = None
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except retry_on as exc: # type: ignore[misc]
last_exception = exc
delay = base_delay * (2 ** (attempt - 1))
LOGGER.warning(
"Попытка %d/%d операции %s не удалась: %s; повтор через %.4f с",
attempt,
max_attempts,
func.__qualname__,
exc,
delay,
)
time.sleep(delay)
raise RetryBudgetExhaustedError(
"Бюджет повторов исчерпан",
details={"attempts": max_attempts, "last_error": str(last_exception)},
) from last_exception
return wrapper
return decorator
def lru_cached(maxsize: int = 128) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""Декоратор кэширования с явно указанным размером кэша."""
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
return functools.lru_cache(maxsize=maxsize)(func)
return decorator
# =============================================================================
# РАЗДЕЛ 16. Контейнер инверсии зависимостей
# =============================================================================
T = TypeVar("T")
@final
@dataclasses.dataclass(frozen=True)
class ServiceDescriptor:
"""Описание сервиса: как его создать и как долго он живёт."""
interface: type
factory: Callable[["ServiceContainer"], Any]
lifetime: ServiceLifetime
class ServiceContainer:
"""Миниатюрный контейнер внедрения зависимостей с поддержкой жизненных циклов.
Возможности:
* регистрация готовых singleton-инстансов;
* регистрация фабрик (singleton / transient);
* обнаружение циклических зависимостей;
* потокобезопасная резолюция.
"""
def __init__(self) -> None:
self._descriptors: Dict[type, ServiceDescriptor] = {}
self._instances: Dict[type, Any] = {}
self._resolution_stack: List[type] = []
self._lock: threading.RLock = threading.RLock()
def register_instance(self, interface: Type[T], instance: T) -> "ServiceContainer":
"""Регистрирует готовый экземпляр как singleton."""
with self._lock:
self._instances[interface] = instance
LOGGER.debug("Зарегистрирован инстанс для %s", interface.__name__)
return self
def register_singleton(
self,
interface: Type[T],
factory: Callable[["ServiceContainer"], T],
) -> "ServiceContainer":
"""Регистрирует фабрику singleton-сервиса."""
return self._register(interface, factory, ServiceLifetime.SINGLETON)
def register_transient(
self,
interface: Type[T],
factory: Callable[["ServiceContainer"], T],
) -> "ServiceContainer":
"""Регистрирует фабрику transient-сервиса."""
return self._register(interface, factory, ServiceLifetime.TRANSIENT)
def _register(
self,
interface: type,
factory: Callable[["ServiceContainer"], Any],
lifetime: ServiceLifetime,
) -> "ServiceContainer":
if not isinstance(interface, type):
raise ConfigurationError("Интерфейс должен быть классом")
with self._lock:
self._descriptors[interface] = ServiceDescriptor(interface, factory, lifetime)
LOGGER.debug(
"Зарегистрирован сервис %s (время жизни: %s)",
interface.__name__,
lifetime.value,
)
return self
def resolve(self, interface: Type[T]) -> T:
"""Резолвит сервис, рекурсивно разрешая его зависимости."""
with self._lock:
return cast(T, self._resolve_locked(interface))
def _resolve_locked(self, interface: type) -> Any:
if interface in self._instances:
return self._instances[interface]
descriptor = self._descriptors.get(interface)
if descriptor is None:
raise DependencyNotFoundError(
"Сервис не зарегистрирован в контейнере",
details={"service": getattr(interface, "__name__", str(interface))},
)
if interface in self._resolution_stack:
chain = [item.__name__ for item in self._resolution_stack]
chain.append(interface.__name__)
raise CircularDependencyError(
"Обнаружена циклическая зависимость",
details={"chain": " -> ".join(chain)},
)
self._resolution_stack.append(interface)
try:
instance = descriptor.factory(self)
if descriptor.lifetime is ServiceLifetime.SINGLETON:
self._instances[interface] = instance
return instance
finally:
self._resolution_stack.pop()
def diagnostics(self) -> Dict[str, Any]:
"""Возвращает статистику контейнера для отладочных целей."""
with self._lock:
return {
"registered": sorted(name.__name__ for name in self._descriptors),
"materialized": sorted(name.__name__ for name in self._instances),
}
# =============================================================================
# РАЗДЕЛ 17. Оркестратор предметной области
# =============================================================================
@final
class HelloWorldApplication:
"""Главный оркестратор: валидирует, рендерит, доставляет и отчитывается."""
def __init__(
self,
provider: GreetingProvider,
strategy: RenderingStrategy,
channel: AbstractOutputChannel,
validators: ValidationChain,
bus: EventBus,
telemetry: TelemetrySink,
clock: Clock,
) -> None:
self._provider: GreetingProvider = provider
self._strategy: RenderingStrategy = strategy
self._channel: AbstractOutputChannel = channel
self._validators: ValidationChain = validators
self._bus: EventBus = bus
self._telemetry: TelemetrySink = telemetry
self._clock: Clock = clock
self._state: LifecycleState = LifecycleState.CREATED
self._runs: int = 0
LOGGER.debug("Приложение %s создано", type(self).__name__)
# -- Инкапсуляция состояния -------------------------------------------------
@property
def state(self) -> LifecycleState:
"""Текущий этап жизненного цикла приложения."""
return self._state
@property
def runs(self) -> int:
"""Число успешно выполненных запусков."""
return self._runs
def _transition(self, target: LifecycleState) -> None:
"""Выполняет переход состояния с проверкой легальности."""
if not self._state.allows_transition_to(target):
raise InvalidStateTransitionError(
"Нелегальный переход жизненного цикла",
details={"from": self._state.value, "to": target.value},
)
LOGGER.debug("Переход состояния: %s -> %s", self._state.value, target.value)
self._state = target
# -- Основной сценарий -------------------------------------------------------
@traced
@retry(max_attempts=3, base_delay=0.0005)
def run(self, request: RenderRequest) -> RenderResult:
"""Исполняет полный сценарий доставки приветствия."""
if self._state is LifecycleState.CREATED:
self._transition(LifecycleState.INITIALIZED)
self._transition(LifecycleState.RUNNING)
self._runs += 1
with trace_context(request.correlation_id):
started = self._clock.monotonic()
try:
greeting = self._provider.provide()
self._validators.validate(greeting)
actual_request = dataclasses.replace(request, greeting=greeting)
payload = self._strategy.render(actual_request)
self._channel.emit(payload)
self._channel.flush()
duration_ms = (self._clock.monotonic() - started) * 1000.0
result = RenderResult(
payload=payload,
byte_length=len(payload.encode("utf-8")),
checksum=hashlib.sha256(payload.encode("utf-8")).hexdigest(),
duration_ms=duration_ms,
correlation_id=request.correlation_id,
)
except Exception as exc:
self._transition(LifecycleState.FAILED)
self._bus.publish(
Event(
name="app.failed",
payload={
"error": str(exc),
"correlation_id": request.correlation_id,
},
)
)
self._telemetry.record("run.failures", 1.0, {"provider": type(self._provider).__name__})
raise
self._transition(LifecycleState.COMPLETED)
self._bus.publish(
Event(
name="message.emitted",
payload={
"duration_ms": duration_ms,
"byte_length": result.byte_length,
"channel": self._channel.name,
},
)
)
self._telemetry.record(
"run.duration_ms",
duration_ms,
{"provider": type(self._provider).__name__, "channel": self._channel.name},
)
LOGGER.info(
"Приветствие доставлено: %s (%d байт, %.3f мс)",
result.payload,
result.byte_length,
result.duration_ms,
)
return result
# -- Контекстное управление --------------------------------------------------
@contextlib.contextmanager
def session(self) -> Iterable["HelloWorldApplication"]:
"""Контекстный менеджер, гарантирующий закрытие канала вывода."""
try:
yield self
finally:
self._channel.close()
LOGGER.debug("Сессия приложения завершена")
# =============================================================================
# РАЗДЕЛ 18. Конфигурация и сборка приложения (composition root)
# =============================================================================
@final
@dataclasses.dataclass(frozen=True)
class ApplicationConfig:
"""Иммутабельная конфигурация приложения."""
provider_key: str = "default"
message_format: MessageFormat = MessageFormat.PLAIN
channel_type: OutputChannelType = OutputChannelType.STDOUT
verbosity: Verbosity = Verbosity.NORMAL
output_json: bool = False
def configure_logging(verbosity: Verbosity) -> None:
"""Настраивает корневой логгер модуля в зависимости от детализации."""
levels: Mapping[Verbosity, int] = {
Verbosity.QUIET: logging.ERROR,
Verbosity.NORMAL: logging.WARNING,
Verbosity.VERBOSE: logging.INFO,
Verbosity.DEBUG: logging.DEBUG,
}
handler = logging.StreamHandler(stream=sys.stderr)
handler.setFormatter(
logging.Formatter(
fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
)
LOGGER.handlers.clear()
LOGGER.addHandler(handler)
LOGGER.setLevel(levels[verbosity])
LOGGER.propagate = False
def build_container(config: ApplicationConfig) -> ServiceContainer:
"""Собирает контейнер зависимостей (composition root)."""
container = ServiceContainer()
container.register_instance(Clock, SystemClock())
container.register_instance(TelemetrySink, InMemoryTelemetrySink())
container.register_singleton(ValidationChain, lambda c: build_default_validation_chain())
container.register_singleton(
GreetingProvider,
lambda c: ProviderRegistryMeta.get(config.provider_key)(),
)
container.register_singleton(
MessageFormatter,
lambda c: create_formatter(config.message_format),
)
container.register_singleton(
RenderingStrategy,
lambda c: CompositeRenderingStrategy(c.resolve(MessageFormatter)),
)
container.register_singleton(
AbstractOutputChannel,
lambda c: create_channel(config.channel_type),
)
container.register_singleton(EventBus, lambda c: EventBus(c.resolve(Clock)))
return container
def build_application(container: ServiceContainer) -> HelloWorldApplication:
"""Создаёт оркестратор, извлекая все зависимости из контейнера."""
bus = container.resolve(EventBus)
telemetry = container.resolve(TelemetrySink)
clock = container.resolve(Clock)
bus.subscribe("message.emitted", metric_recorder(telemetry, clock))
bus.subscribe("app.failed", metric_recorder(telemetry, clock))
if LOGGER.isEnabledFor(logging.INFO):
bus.subscribe("message.emitted", console_announcer)
return HelloWorldApplication(
provider=container.resolve(GreetingProvider),
strategy=container.resolve(RenderingStrategy),
channel=container.resolve(AbstractOutputChannel),
validators=container.resolve(ValidationChain),
bus=bus,
telemetry=telemetry,
clock=clock,
)
def build_request(config: ApplicationConfig) -> RenderRequest:
"""Формирует запрос рендеринга из конфигурации."""
return RenderRequest(
greeting=Greeting(),
message_format=config.message_format,
verbosity=config.verbosity,
)
# =============================================================================
# РАЗДЕЛ 19. Самопроверочный набор тестов (self-test)
# =============================================================================
class SelfTestSuite:
"""Набор проверок, подтверждающих работоспособность без фреймворков."""
def __init__(self) -> None:
self._passed: int = 0
self._failed: int = 0
def check(self, condition: bool, name: str) -> None:
"""Фиксирует результат одной проверки."""
if condition:
self._passed += 1
print(f" [OK] {name}")
else:
self._failed += 1
print(f" [FAIL] {name}")
def run_all(self) -> int:
"""Исполняет все проверки и возвращает код завершения."""
print("Запуск самопроверочного набора...")
previous_level = logging.root.manager.disable
logging.disable(logging.CRITICAL) # тесты намеренно провоцируют отказы
try:
self.test_greeting_composition()
self.test_greeting_invariants()
self.test_formatters()
self.test_validators()
self.test_registry()
self.test_container_circular_detection()
self.test_event_bus_isolation()
self.test_buffer_channel()
self.test_end_to_end()
finally:
logging.disable(previous_level)
print(f"Итог: пройдено {self._passed}, провалено {self._failed}")
return 0 if self._failed == 0 else 1
def test_greeting_composition(self) -> None:
greeting = Greeting()
self.check(greeting.compose() == "Hello, World!", "Канонический состав: Hello, World!")
def test_greeting_invariants(self) -> None:
try:
Greeting(salutation=" ")
self.check(False, "Пустая salutation отклоняется")
except ValidationError:
self.check(True, "Пустая salutation отклоняется")
def test_formatters(self) -> None:
self.check(create_formatter(MessageFormat.UPPER).format("ab") == "AB", "Верхний регистр")
self.check(create_formatter(MessageFormat.LOWER).format("AB") == "ab", "Нижний регистр")
self.check(create_formatter(MessageFormat.PLAIN).format("x") == "x", "Неизменный текст")
def test_validators(self) -> None:
chain = build_default_validation_chain()
try:
chain.validate(Greeting())
self.check(True, "Валидация корректного приветствия")
except ValidationError:
self.check(False, "Валидация корректного приветствия")
def test_registry(self) -> None:
keys = ProviderRegistryMeta.keys()
self.check("default" in keys, "Реестр содержит провайдер default")
self.check(ProviderRegistryMeta.get("default") is DefaultGreetingProvider, "Ключ default резолвится")
def test_container_circular_detection(self) -> None:
class A:
def __init__(self, container: "ServiceContainer") -> None:
self.b = container.resolve("B") # type: ignore[arg-type]
container = ServiceContainer()
container.register_singleton(A, lambda c: A(c))
try:
container.resolve(A)
self.check(False, "Циклические зависимости обнаруживаются")
except (CircularDependencyError, DependencyNotFoundError):
self.check(True, "Циклические зависимости обнаруживаются")
def test_event_bus_isolation(self) -> None:
bus = EventBus(SystemClock())
bus.subscribe("x", lambda e: (_ for _ in ()).throw(RuntimeError("boom")))
delivered = bus.publish(Event(name="x"))
self.check(delivered == 0, "Отказ подписчика не прерывает публикацию")
def test_buffer_channel(self) -> None:
channel = BufferChannel()
channel.emit("ping")
self.check(channel.drain() == ("ping",), "Буферный канал накапливает сообщения")
def test_end_to_end(self) -> None:
config = ApplicationConfig(channel_type=OutputChannelType.BUFFER)
container = build_container(config)
application = build_application(container)
result = application.run(build_request(config))
channel = container.resolve(AbstractOutputChannel)
buffer = cast(BufferChannel, channel)
self.check(result.payload == "Hello, World!", "Сквозной сценарий печатает Hello, World!")
self.check(buffer.drain() == ("Hello, World!",), "Канал получил ровно одно сообщение")
self.check(len(result.checksum) == 64, "Контрольная сумма SHA-256 вычислена")
# =============================================================================
# РАЗДЕЛ 20. Интерфейс командной строки и точка входа
# =============================================================================
def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
"""Разбирает аргументы командной строки."""
parser = argparse.ArgumentParser(
prog="hello_world_overengineered",
description="Самый заинжиниренный вывод Hello, World! в истории.",
)
parser.add_argument("--provider", default="default", help="Ключ провайдера приветствия")
parser.add_argument(
"--format",
dest="message_format",
default="plain",
choices=[item.value for item in MessageFormat],
help="Стратегия форматирования",
)
parser.add_argument(
"--channel",
default="stdout",
choices=[item.value for item in OutputChannelType],
help="Канал вывода",
)
parser.add_argument("--json", action="store_true", help="Вывод результата в формате JSON")
parser.add_argument("-v", "--verbose", action="count", default=0, help="Уровень детализации")
parser.add_argument("--self-test", action="store_true", help="Запустить самопроверку")
return parser.parse_args(argv)
def config_from_args(args: argparse.Namespace) -> ApplicationConfig:
"""Преобразует разобранные аргументы в типизированную конфигурацию."""
verbosity = Verbosity.NORMAL + min(args.verbose, len(Verbosity) - 1)
return ApplicationConfig(
provider_key=args.provider,
message_format=MessageFormat(args.message_format),
channel_type=OutputChannelType(args.channel),
verbosity=Verbosity(verbosity),
output_json=bool(args.json),
)
def main(argv: Optional[Sequence[str]] = None) -> int:
"""Точка входа приложения."""
args = parse_args(argv)
if args.self_test:
return SelfTestSuite().run_all()
config = config_from_args(args)
if config.output_json and config.channel_type is OutputChannelType.STDOUT:
# В JSON-режиме stdout принадлежит исключительно структурированному
# результату, поэтому сообщение уходит в буферный канал.
config = dataclasses.replace(config, channel_type=OutputChannelType.BUFFER)
configure_logging(config.verbosity)
LOGGER.info("Инициализация приложения, версия %s", __version__)
container = build_container(config)
application = build_application(container)
atexit.register(lambda: LOGGER.info("Контейнер: %s", container.diagnostics()))
try:
with application.session():
result = application.run(build_request(config))
except HelloWorldError as exc:
print(json.dumps(exc.to_dict(), ensure_ascii=False), file=sys.stderr)
return 2
except Exception:
traceback.print_exc()
return 70
# Сообщение уже доставлено каналом вывода, поэтому здесь мы лишь
# отчитываемся о метаданных рендеринга, если об этом просили.
if config.output_json:
print(json.dumps(result.to_dict(), ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())
