Протоколы в Python

from abc import ABC, abstractmethod
 
class Serializable(ABC):
    @abstractmethod
    def to_json(self) -> str:
        pass
    
    @abstractmethod
    def from_json(self, data: str):
        pass
 
class User(Serializable):  # Явное наследование обязательно
    def __init__(self, name: str):
        self.name = name
    
    def to_json(self) -> str:
        return f'{{"name": "{self.name}"}}'
    
    def from_json(self, data: str):
        # Разбор JSON
        pass

from typing import Protocol
 
class Serializable(Protocol):
    def to_json(self) -> str: ...
    def from_json(self, data: str): ...
 
class User:  # Никакого наследования!
    def __init__(self, name: str):
        self.name = name
    
    def to_json(self) -> str:
        return f'{{"name": "{self.name}"}}'
    
    def from_json(self, data: str):
        pass
 
def save_to_db(obj: Serializable) -> None:
    # User подходит, потому что реализует нужные методы
    data = obj.to_json()
    # Сохраняем...

def process(obj):
    # Надеемся, что у obj есть метод handle
    obj.handle()  # Узнаем только здесь, есть он или нет

class Drawable(Protocol):
    """Объекты, которые можно нарисовать на холсте"""
    def draw(self, canvas: Canvas) -> None:
        """Отрисовка объекта на переданном холсте"""
        ...
    
    def get_bounds(self) -> tuple[int, int, int, int]:
        """Возвращает границы объекта: (x, y, width, height)"""
        ...

from typing import Protocol

from typing import Protocol
 
class Comparable(Protocol):
    def __lt__(self, other) -> bool:
        """Меньше чем"""
        ...
    
    def __le__(self, other) -> bool:
        """Меньше или равно"""
        ...

class Identifiable(Protocol):
    id: int  # Атрибут, который должен существовать
    name: str
    
    def get_id(self) -> int:
        """Получить идентификатор"""
        ...

class User:
    def __init__(self, user_id: int, username: str):
        self.id = user_id
        self.name = username
    
    def get_id(self) -> int:
        return self.id
 
user = User(42, "admin")
# isinstance(user, Identifiable)  # Ошибка: Protocols cannot be used with isinstance

from typing import Protocol, runtime_checkable
 
@runtime_checkable
class Identifiable(Protocol):
    id: int
    name: str
    
    def get_id(self) -> int: ...
 
# Теперь isinstance работает
user = User(42, "admin")
print(isinstance(user, Identifiable))  # True

@runtime_checkable
class Processor(Protocol):
    def process(self, data: str) -> dict: ...
 
class BrokenProcessor:
    def process(self, x: int, y: int) -> list:  # Другая сигнатура!
        return [x, y]
 
broken = BrokenProcessor()
print(isinstance(broken, Processor))  # True - только проверка наличия метода!

print(issubclass(User, Identifiable))  # True
print(issubclass(BrokenProcessor, Processor))  # True

class Readable(Protocol):
    def read(self) -> bytes: ...
 
class Writable(Protocol):
    def write(self, data: bytes) -> int: ...
 
class ReadWritable(Readable, Writable, Protocol):
    """Объект, который можно читать и записывать"""
    pass

class Sized(Protocol):
    @property
    def size(self) -> int:
        """Размер объекта"""
        ...
 
class Buffer:
    def __init__(self, capacity: int):
        self._capacity = capacity
    
    @property
    def size(self) -> int:
        return self._capacity
 
# Buffer соответствует Sized

class Timestamped(Protocol):
    created_at: float
    
    def age(self) -> float:
        """Возвращает возраст объекта в секундах"""
        import time
        return time.time() - self.created_at

from typing import Protocol, TypeVar
 
T = TypeVar('T')
 
class Container(Protocol[T]):
    def add(self, item: T) -> None: ...
    def get(self) -> T: ...
 
class IntContainer:
    def __init__(self):
        self._items: list[int] = []
    
    def add(self, item: int) -> None:
        self._items.append(item)
    
    def get(self) -> int:
        return self._items[0] if self._items else 0
 
# IntContainer соответствует Container[int]

class Loggable(Protocol):
    def log(self, message: str) -> None: ...
 
# Так нельзя
# logger = Loggable()  # TypeError: Protocols cannot be instantiated

from typing import SupportsInt
 
def process_numeric(value: SupportsInt) -> int:
    # Принимаем всё, что можно превратить в int
    return int(value)
 
class Custom:
    def __int__(self) -> int:
        return 42
 
# Работает
result = process_numeric(Custom())  # 42

class Addable(Protocol):
    def __add__(self, other) -> 'Addable':
        """Поддержка оператора +"""
        ...
 
class Vector2D:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y
    
    def __add__(self, other: 'Vector2D') -> 'Vector2D':
        return Vector2D(self.x + other.x, self.y + other.y)
 
def combine(a: Addable, b: Addable) -> Addable:
    return a + b
 
# Vector2D соответствует Addable
v1 = Vector2D(1, 2)
v2 = Vector2D(3, 4)
v3 = combine(v1, v2)  # Статический анализатор доволен

class Serializer(Protocol):
    class Config(Protocol):
        indent: int
        sort_keys: bool
    
    def serialize(self, obj: object, config: Config) -> str: ...

from typing import Protocol, TypeVar
 
T_co = TypeVar('T_co', covariant=True)  # Ковариантный параметр
T_contra = TypeVar('T_contra', contravariant=True)  # Контравариантный
 
class Producer(Protocol[T_co]):
    def produce(self) -> T_co: ...
 
class Consumer(Protocol[T_contra]):
    def consume(self, item: T_contra) -> None: ...

class BaseClass:
    def method(self) -> None: ...
 
# Так нельзя
# class Mixed(BaseClass, Protocol):  # TypeError
#     def protocol_method(self) -> None: ...

class Countdown:
    def __init__(self, start: int):
        self.current = start
    
    def __iter__(self):
        # Возвращаем итератор
        return CountdownIterator(self.current)
 
class CountdownIterator:
    def __init__(self, start: int):
        self.value = start
    
    def __iter__(self):
        return self
    
    def __next__(self) -> int:
        if self.value <= 0:
            raise StopIteration
        self.value -= 1
        return self.value + 1
 
# Работает в цикле
for num in Countdown(5):
    print(num)  # 5, 4, 3, 2, 1

from typing import Iterator, Iterable
 
def process_items(items: Iterable[int]) -> None:
    for item in items:
        # Обработка каждого элемента
        print(item * 2)
 
# Подходит любой итерируемый объект
process_items([1, 2, 3])  # Список
process_items(range(5))   # Range
process_items(Countdown(3))  # Наш класс

from typing import Protocol, TypeVar
 
T = TypeVar('T')
 
class ContextManager(Protocol[T]):
    def __enter__(self) -> T: ...
    def __exit__(self, exc_type, exc_val, exc_tb) -> bool | None: ...
 
class DatabaseConnection:
    def __init__(self, host: str):
        self.host = host
        self.connected = False
    
    def __enter__(self):
        # Устанавливаем соединение
        self.connected = True
        print(f"Подключились к {self.host}")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Закрываем соединение
        self.connected = False
        print(f"Отключились от {self.host}")
        return False  # Не подавляем исключения
 
def execute_query(manager: ContextManager[DatabaseConnection]) -> None:
    with manager as conn:
        # Работаем с подключением
        assert conn.connected

from typing import Callable
 
class Multiplier:
    def __init__(self, factor: int):
        self.factor = factor
    
    def __call__(self, x: int) -> int:
        # Объект можно вызывать как функцию
        return x * self.factor
 
def apply_operation(value: int, operation: Callable[[int], int]) -> int:
    return operation(value)
 
doubler = Multiplier(2)
result = apply_operation(10, doubler)  # 20
result = apply_operation(10, lambda x: x + 5)  # 15

from typing import Sized, Container
 
class CustomSet:
    def __init__(self, items: list[int]):
        self._items = set(items)
    
    def __len__(self) -> int:
        return len(self._items)
    
    def __contains__(self, item: int) -> bool:
        return item in self._items
 
def check_size(obj: Sized) -> bool:
    # Работает с любым объектом, у которого есть len()
    return len(obj) > 0
 
def check_membership(container: Container[int], value: int) -> bool:
    return value in container
 
custom = CustomSet([1, 2, 3])
print(check_size(custom))  # True
print(check_membership(custom, 2))  # True

class Descriptor(Protocol):
    def __get__(self, instance, owner): ...
    def __set__(self, instance, value): ...
    def __delete__(self, instance): ...
 
class ValidatedAttribute:
    def __init__(self, min_value: int):
        self.min_value = min_value
        self.value = 0
    
    def __get__(self, instance, owner):
        return self.value
    
    def __set__(self, instance, value: int):
        if value < self.min_value:
            raise ValueError(f"Значение должно быть >= {self.min_value}")
        self.value = value
 
class Product:
    price = ValidatedAttribute(0)  # Цена не может быть отрицательной

from dataclasses import dataclass
from typing import Protocol
 
class Comparable(Protocol):
    def __lt__(self, other) -> bool: ...
    def __le__(self, other) -> bool: ...
 
@dataclass(order=True)  # Генерирует методы сравнения
class Score:
    value: int
    name: str
 
def get_max(a: Comparable, b: Comparable) -> Comparable:
    return a if a > b else b
 
# Score автоматически соответствует Comparable
score1 = Score(100, "Alice")
score2 = Score(85, "Bob")
winner = get_max(score1, score2)  # score1

import attr
from typing import Protocol
 
class Hashable(Protocol):
    def __hash__(self) -> int: ...
    def __eq__(self, other) -> bool: ...
 
@attr.s(frozen=True, auto_attribs=True)  # frozen генерирует __hash__
class UserId:
    value: int
    
    def validate(self):
        if self.value <= 0:
            raise ValueError("ID должен быть положительным")
 
# UserId соответствует Hashable
ids = {UserId(1), UserId(2), UserId(1)}  # Работает как множество

from typing import Reversible, Iterator
 
class Playlist:
    def __init__(self, tracks: list[str]):
        self._tracks = tracks
    
    def __iter__(self) -> Iterator[str]:
        return iter(self._tracks)
    
    def __reversed__(self) -> Iterator[str]:
        # Итерация в обратном порядке
        return reversed(self._tracks)
 
def shuffle_backwards(items: Reversible[str]) -> None:
    for item in reversed(items):
        print(f"Проигрываем: {item}")
 
playlist = Playlist(["Intro", "Main", "Outro"])
shuffle_backwards(playlist)  # Outro, Main, Intro

from typing import Mapping, MutableMapping, Iterator
 
class ConfigStore:
    def __init__(self):
        self._data: dict[str, str] = {}
    
    def __getitem__(self, key: str) -> str:
        return self._data[key]
    
    def __setitem__(self, key: str, value: str) -> None:
        # Валидация при записи
        if not key.isupper():
            raise KeyError("Ключи должны быть в верхнем регистре")
        self._data[key] = value
    
    def __delitem__(self, key: str) -> None:
        del self._data[key]
    
    def __iter__(self) -> Iterator[str]:
        return iter(self._data)
    
    def __len__(self) -> int:
        return len(self._data)
 
def read_config(config: Mapping[str, str]) -> str:
    # Только чтение, изменения невозможны
    return config.get("DATABASE_URL", "localhost")
 
def update_config(config: MutableMapping[str, str], key: str, value: str) -> None:
    # Можно изменять
    config[key] = value
 
store = ConfigStore()
update_config(store, "API_KEY", "secret123")  # Ошибка: ключ не в верхнем регистре

from typing import Sequence, MutableSequence
 
class RingBuffer:
    def __init__(self, capacity: int):
        self._capacity = capacity
        self._data: list[int] = []
    
    def __getitem__(self, index: int) -> int:
        # Циклическая индексация
        return self._data[index % len(self._data)]
    
    def __setitem__(self, index: int, value: int) -> None:
        self._data[index % len(self._data)] = value
    
    def __len__(self) -> int:
        return len(self._data)
    
    def append(self, value: int) -> None:
        if len(self._data) < self._capacity:
            self._data.append(value)
        else:
            # Перезаписываем старейший элемент
            self._data[len(self._data) % self._capacity] = value
 
def sum_sequence(seq: Sequence[int]) -> int:
    # Работает с любой последовательностью
    total = 0
    for val in seq:
        total += val
    return total
 
buffer = RingBuffer(5)
buffer.append(10)
buffer.append(20)
print(sum_sequence(buffer))  # 30

from typing import AsyncIterator
import asyncio
 
class AsyncCounter:
    def __init__(self, limit: int):
        self.limit = limit
        self.current = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self) -> int:
        if self.current >= self.limit:
            raise StopAsyncIteration
        # Имитация асинхронной операции
        await asyncio.sleep(0.1)
        self.current += 1
        return self.current
 
async def process_async(source: AsyncIterator[int]) -> None:
    async for value in source:
        print(f"Обработали: {value}")
 
# asyncio.run(process_async(AsyncCounter(3)))

from typing import SupportsAbs, SupportsRound
 
class ComplexNumber:
    def __init__(self, real: float, imag: float):
        self.real = real
        self.imag = imag
    
    def __abs__(self) -> float:
        # Модуль комплексного числа
        return (self.real [B] 2 + self.imag [/B] 2) ** 0.5
    
    def __round__(self, ndigits: int = 0) -> 'ComplexNumber':
        return ComplexNumber(
            round(self.real, ndigits),
            round(self.imag, ndigits)
        )
 
def calculate_distance(point: SupportsAbs[float]) -> float:
    # abs() работает с любым объектом, поддерживающим __abs__
    return abs(point)
 
num = ComplexNumber(3.0, 4.0)
print(calculate_distance(num))  # 5.0

from typing import Protocol
 
class NotificationChannel(Protocol):
    def send(self, recipient: str, message: str) -> bool:
        """Отправка сообщения получателю. Возвращает True при успехе"""
        ...
    
    def is_available(self) -> bool:
        """Проверка доступности канала"""
        ...
 
class EmailNotifier:
    def __init__(self, smtp_host: str, smtp_port: int):
        self.host = smtp_host
        self.port = smtp_port
        self.connected = False
    
    def send(self, recipient: str, message: str) -> bool:
        # Отправка через SMTP
        if not self.connected:
            return False
        print(f"Email на {recipient}: {message}")
        return True
    
    def is_available(self) -> bool:
        return self.connected
 
class TelegramNotifier:
    def __init__(self, bot_token: str):
        self.token = bot_token
    
    def send(self, recipient: str, message: str) -> bool:
        # Отправка через Telegram API
        print(f"Telegram для {recipient}: {message}")
        return True
    
    def is_available(self) -> bool:
        return bool(self.token)
 
def broadcast(channels: list[NotificationChannel], users: list[str], text: str) -> None:
    """Рассылка сообщения через все доступные каналы"""
    for channel in channels:
        if not channel.is_available():
            continue
        for user in users:
            channel.send(user, text)

from typing import Protocol
 
class Serializable(Protocol):
    def to_dict(self) -> dict:
        """Преобразование объекта в словарь"""
        ...
 
class Validatable(Protocol):
    def validate(self) -> bool:
        """Проверка корректности данных"""
        ...
 
class Persistable(Serializable, Validatable, Protocol):
    """Объект, который можно сериализовать, валидировать и сохранять"""
    def save_to_db(self) -> None:
        """Сохранение в базу данных"""
        ...
 
class UserProfile:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age
    
    def to_dict(self) -> dict:
        return {"name": self.name, "age": self.age}
    
    def validate(self) -> bool:
        # Простая валидация
        return len(self.name) > 0 and self.age >= 0
    
    def save_to_db(self) -> None:
        if not self.validate():
            raise ValueError("Некорректные данные профиля")
        # Сохранение в базу
        data = self.to_dict()
        print(f"Сохраняем: {data}")
 
def persist_entity(entity: Persistable) -> None:
    """Универсальная функция сохранения"""
    if entity.validate():
        entity.save_to_db()
        print(f"Сохранено: {entity.to_dict()}")
    else:
        print("Валидация провалилась")
 
profile = UserProfile("Alice", 30)
persist_entity(profile)

from typing import Protocol, TypeVar, Generic
 
T = TypeVar('T')
R = TypeVar('R')
 
class Mapper(Protocol[T, R]):
    """Преобразование объектов одного типа в другой"""
    def map(self, item: T) -> R:
        ...
 
class IntToStringMapper:
    def map(self, item: int) -> str:
        # Преобразуем число в строку с форматированием
        return f"Число: {item:05d}"
 
class UserToJsonMapper:
    def map(self, item: dict) -> str:
        # Преобразуем словарь в JSON-строку
        import json
        return json.dumps(item)
 
def transform_batch(items: list[T], mapper: Mapper[T, R]) -> list[R]:
    """Применение преобразования к списку элементов"""
    return [mapper.map(item) for item in items]
 
# Работает с разными типами
int_mapper = IntToStringMapper()
strings = transform_batch([1, 2, 3], int_mapper)  # ["Число: 00001", ...]
 
user_mapper = UserToJsonMapper()
users = [{"name": "Alice"}, {"name": "Bob"}]
jsons = transform_batch(users, user_mapper)  # ['{"name": "Alice"}', ...]

from typing import Protocol, TypeVar
 
T_co = TypeVar('T_co', covariant=True)
T_contra = TypeVar('T_contra', contravariant=True)
 
class Source(Protocol[T_co]):
    """Источник данных - только чтение"""
    def get(self) -> T_co:
        ...
 
class Sink(Protocol[T_contra]):
    """Приёмник данных - только запись"""
    def put(self, item: T_contra) -> None:
        ...
 
class IntSource:
    def get(self) -> int:
        return 42
 
class NumberSink:
    def put(self, item: int | float) -> None:
        print(f"Получили число: {item}")
 
# Source ковариантен: Source[int] является подтипом Source[int | float]
# Sink контравариантен: Sink[int | float] является подтипом Sink[int]

from typing import Protocol
 
class Measurable(Protocol):
    @property
    def width(self) -> float:
        """Ширина объекта"""
        ...
    
    @property
    def height(self) -> float:
        """Высота объекта"""
        ...
 
class Rectangle:
    def __init__(self, w: float, h: float):
        self._width = w
        self._height = h
    
    @property
    def width(self) -> float:
        return self._width
    
    @property
    def height(self) -> float:
        return self._height
 
class Circle:
    def __init__(self, radius: float):
        self._radius = radius
    
    @property
    def width(self) -> float:
        # Диаметр круга
        return self._radius * 2
    
    @property
    def height(self) -> float:
        return self._radius * 2
 
def calculate_area(obj: Measurable) -> float:
    """Приблизительная площадь"""
    return obj.width * obj.height
 
rect = Rectangle(10, 5)
circ = Circle(3)
print(calculate_area(rect))  # 50.0
print(calculate_area(circ))  # 36.0

from typing import Protocol
 
class Comparable(Protocol):
    def __lt__(self, other) -> bool:
        """Оператор <"""
        ...
    
    def __le__(self, other) -> bool:
        """Оператор <="""
        ...
 
class Priority:
    def __init__(self, level: int, name: str):
        self.level = level
        self.name = name
    
    def __lt__(self, other: 'Priority') -> bool:
        # Сравнение по уровню приоритета
        return self.level < other.level
    
    def __le__(self, other: 'Priority') -> bool:
        return self.level <= other.level
 
def find_highest(items: list[Comparable]) -> Comparable:
    """Поиск элемента с максимальным приоритетом"""
    if not items:
        raise ValueError("Список пуст")
    highest = items[0]
    for item in items[1:]:
        if item > highest:  # Использует __lt__ и __le__
            highest = item
    return highest
 
tasks = [Priority(3, "Low"), Priority(1, "Critical"), Priority(2, "Medium")]
urgent = find_highest(tasks)
print(urgent.name)  # "Critical"

from typing import Protocol
 
class Repository(Protocol):
    def find(self, id: int) -> dict | None:
        """Поиск записи по ID"""
        ...
    
    def save(self, data: dict) -> int:
        """Сохранение и возврат ID"""
        ...
 
class UnitOfWork(Protocol):
    repository: Repository  # Протокол как атрибут!
    
    def commit(self) -> None:
        """Фиксация транзакции"""
        ...
    
    def rollback(self) -> None:
        """Откат изменений"""
        ...

mypy --strict --show-error-codes src/

from typing import Protocol
 
class Logger(Protocol):
    def log(self, message: str, level: str) -> None:
        ...
 
def process_data(logger: Logger, data: list[int]) -> None:
    logger.log(f"Обработка {len(data)} элементов", "INFO")
    # Обработка данных
    logger.log("Готово", "INFO")
 
class FileLogger:
    def __init__(self, path: str):
        self.path = path
    
    def log(self, message: str) -> None:  # Неправильная сигнатура!
        # Записываем в файл
        with open(self.path, 'a') as f:
            f.write(message + '\n')
 
# mypy увидит ошибку
file_logger = FileLogger("app.log")
process_data(file_logger, [1, 2, 3])  # error: Missing positional argument "level"

from typing import Protocol
 
class Storage(Protocol):
    capacity: int  # Атрибут обязателен
    
    def store(self, item: str) -> bool:
        ...
    
    def retrieve(self) -> str | None:
        ...
 
class MemoryStorage:
    def __init__(self):
        self.data: list[str] = []
        # Забыли capacity!
    
    def store(self, item: str) -> bool:
        self.data.append(item)
        return True
    
    def retrieve(self) -> str | None:
        return self.data.pop() if self.data else None
 
def use_storage(storage: Storage) -> None:
    print(f"Ёмкость: {storage.capacity}")  # mypy: error: "MemoryStorage" has no attribute "capacity"

# Было
def send_notification(channel, user, message):
    channel.send(user, message)
 
# Стало
from typing import Protocol
 
class NotificationChannel(Protocol):
    def send(self, recipient: str, message: str) -> bool:
        ...
 
def send_notification(channel: NotificationChannel, user: str, message: str) -> None:
    channel.send(user, message)

mypy --no-strict-optional src/notifications.py

[mypy]
python_version = 3.11
warn_return_any = True
warn_unused_configs = True
disallow_untyped_defs = False  # Для legacy кода
 
[mypy-tests.*]
ignore_errors = True  # Игнорируем тесты пока
 
[mypy-third_party.*]
ignore_missing_imports = True  # Сторонние библиотеки без типов

from typing import Protocol, TypeVar
 
T = TypeVar('T')
 
class Container(Protocol[T]):
    def get(self) -> T: ...
 
def process(c: Container) -> None:  # Ошибка! Нужно Container[SomeType]
    pass
 
def process_correct(c: Container[int]) -> None:  # Правильно
    item = c.get()  # mypy знает, что item: int

class Reader(Protocol):
    def read(self) -> str | None: ...
 
class StrictReader:
    def read(self) -> str:  # Всегда возвращает строку - OK
        return "data"
 
class OptionalReader:
    def read(self) -> str | None:  # Может вернуть None - тоже OK
        return None if random.random() < 0.5 else "data"

mypy --cache-dir=.mypy_cache --incremental src/

from typing import TYPE_CHECKING, Protocol
 
if TYPE_CHECKING:
    from module_b import SomeClass  # Импорт только для проверки типов
 
class MyProtocol(Protocol):
    def method(self, obj: 'SomeClass') -> None: ...  # Строковая аннотация

from typing import Protocol
from decimal import Decimal
 
class PaymentGateway(Protocol):
    def charge(self, amount: Decimal, currency: str, card_token: str) -> str:
        """Списание средств. Возвращает ID транзакции"""
        ...
    
    def refund(self, transaction_id: str, amount: Decimal) -> bool:
        """Возврат средств"""
        ...
    
    def get_status(self, transaction_id: str) -> str:
        """Проверка статуса платежа"""
        ...
 
class StripeAdapter:
    def __init__(self, api_key: str):
        import stripe
        stripe.api_key = api_key
        self.stripe = stripe
    
    def charge(self, amount: Decimal, currency: str, card_token: str) -> str:
        # Stripe работает с копейками
        amount_cents = int(amount * 100)
        charge = self.stripe.Charge.create(
            amount=amount_cents,
            currency=currency.lower(),
            source=card_token
        )
        return charge.id
    
    def refund(self, transaction_id: str, amount: Decimal) -> bool:
        try:
            refund = self.stripe.Refund.create(
                charge=transaction_id,
                amount=int(amount * 100)
            )
            return refund.status == 'succeeded'
        except Exception:
            return False
    
    def get_status(self, transaction_id: str) -> str:
        charge = self.stripe.Charge.retrieve(transaction_id)
        return charge.status
 
class TinkoffAdapter:
    def __init__(self, terminal_key: str, secret_key: str):
        self.terminal = terminal_key
        self.secret = secret_key
        self.base_url = "https://securepay.tinkoff.ru/v2/"
    
    def charge(self, amount: Decimal, currency: str, card_token: str) -> str:
        # Тинькофф работает с рублями в копейках
        import requests
        response = requests.post(
            f"{self.base_url}Init",
            json={
                "TerminalKey": self.terminal,
                "Amount": int(amount * 100),
                "OrderId": self._generate_order_id(),
                "DATA": {"Token": card_token}
            }
        )
        return response.json()["PaymentId"]
    
    def refund(self, transaction_id: str, amount: Decimal) -> bool:
        # Реализация возврата для Тинькофф
        return True
    
    def get_status(self, transaction_id: str) -> str:
        # Проверка статуса
        return "succeeded"
    
    def _generate_order_id(self) -> str:
        import uuid
        return str(uuid.uuid4())

class MockPaymentGateway:
    def __init__(self):
        self.charges: dict[str, Decimal] = {}
        self.refunds: dict[str, Decimal] = {}
        self._counter = 0
    
    def charge(self, amount: Decimal, currency: str, card_token: str) -> str:
        self._counter += 1
        tx_id = f"mock_tx_{self._counter}"
        self.charges[tx_id] = amount
        return tx_id
    
    def refund(self, transaction_id: str, amount: Decimal) -> bool:
        if transaction_id in self.charges:
            self.refunds[transaction_id] = amount
            return True
        return False
    
    def get_status(self, transaction_id: str) -> str:
        return "succeeded" if transaction_id in self.charges else "failed"
 
def test_payment_flow():
    gateway: PaymentGateway = MockPaymentGateway()
    
    # Тестируем без реальных API-вызовов
    tx_id = gateway.charge(Decimal("100.50"), "RUB", "test_token")
    assert gateway.get_status(tx_id) == "succeeded"
    assert gateway.refund(tx_id, Decimal("50.00"))

from typing import Protocol, Iterator
 
class DataSource(Protocol):
    def fetch(self) -> Iterator[dict]:
        """Получение данных построчно"""
        ...
    
    def close(self) -> None:
        """Очистка ресурсов"""
        ...
 
class CSVSource:
    def __init__(self, filepath: str):
        self.filepath = filepath
        self.file = None
    
    def fetch(self) -> Iterator[dict]:
        import csv
        self.file = open(self.filepath, 'r', encoding='utf-8')
        reader = csv.DictReader(self.file)
        for row in reader:
            yield row
    
    def close(self) -> None:
        if self.file:
            self.file.close()
 
class PostgresSource:
    def __init__(self, connection_string: str):
        self.conn_string = connection_string
        self.connection = None
    
    def fetch(self) -> Iterator[dict]:
        import psycopg2
        import psycopg2.extras
        self.connection = psycopg2.connect(self.conn_string)
        cursor = self.connection.cursor(cursor_factory=psycopg2.extras.DictCursor)
        cursor.execute("SELECT * FROM raw_data")
        for row in cursor:
            yield dict(row)
    
    def close(self) -> None:
        if self.connection:
            self.connection.close()

from typing import Protocol
 
class UserRepository(Protocol):
    def get_by_id(self, user_id: int) -> dict | None: ...
    def save(self, user: dict) -> None: ...
 
class EmailService(Protocol):
    def send(self, recipient: str, subject: str, body: str) -> None: ...
 
class UserService:
    def __init__(self, repo: UserRepository, email: EmailService):
        self.repo = repo
        self.email = email
    
    def register_user(self, username: str, email_addr: str) -> None:
        user = {"username": username, "email": email_addr}
        self.repo.save(user)
        self.email.send(
            email_addr,
            "Добро пожаловать!",
            f"Привет, {username}!"
        )
 
# Реализации можно менять без изменения UserService
class SQLRepository:
    def get_by_id(self, user_id: int) -> dict | None:
        # SQL запрос
        return {"id": user_id, "username": "test"}
    
    def save(self, user: dict) -> None:
        # INSERT запрос
        pass
 
class SMTPEmail:
    def send(self, recipient: str, subject: str, body: str) -> None:
        # Отправка через SMTP
        print(f"Email to {recipient}: {subject}")

# domain/models.py - чистая бизнес-логика
class Order:
    def __init__(self, order_id: int, items: list[dict]):
        self.id = order_id
        self.items = items
        self.total = sum(item['price'] for item in items)
    
    def apply_discount(self, percent: int) -> None:
        self.total *= (1 - percent / 100)
 
# application/protocols.py - контракты для инфраструктуры
from typing import Protocol
 
class OrderRepository(Protocol):
    def find(self, order_id: int) -> Order | None: ...
    def save(self, order: Order) -> None: ...
 
# application/services.py - бизнес-логика
class OrderService:
    def __init__(self, repo: OrderRepository):
        self.repo = repo
    
    def apply_promo_code(self, order_id: int, code: str) -> bool:
        order = self.repo.find(order_id)
        if not order:
            return False
        
        if code == "SALE20":
            order.apply_discount(20)
            self.repo.save(order)
            return True
        return False
 
# infrastructure/repositories.py - детали реализации
class PostgresOrderRepository:
    def __init__(self, db_url: str):
        self.db_url = db_url
    
    def find(self, order_id: int) -> Order | None:
        # SQL запрос и маппинг в Order
        return Order(order_id, [])
    
    def save(self, order: Order) -> None:
        # UPDATE в базе
        pass

from typing import Protocol, TypeVar, Callable, Any
from functools import wraps
 
T = TypeVar('T')
 
class CacheBackend(Protocol):
  def get(self, key: str) -> Any | None:
      """Получение из кеша"""
      ...
  
  def set(self, key: str, value: Any, ttl: int) -> None:
      """Сохранение в кеш с TTL в секундах"""
      ...
 
class MemoryCache:
  def __init__(self):
      self._storage: dict[str, tuple[Any, float]] = {}
  
  def get(self, key: str) -> Any | None:
      import time
      if key in self._storage:
          value, expires = self._storage[key]
          if time.time() < expires:
              return value
          del self._storage[key]
      return None
  
  def set(self, key: str, value: Any, ttl: int) -> None:
      import time
      self._storage[key] = (value, time.time() + ttl)
 
def cached(backend: CacheBackend, ttl: int = 300):
  """Декоратор кеширования с произвольным бэкендом"""
  def decorator(func: Callable[..., T]) -> Callable[..., T]:
      @wraps(func)
      def wrapper(*args, **kwargs) -> T:
          # Генерируем ключ из аргументов
          cache_key = f"{func.__name__}:{args}:{kwargs}"
          
          # Проверяем кеш
          cached_value = backend.get(cache_key)
          if cached_value is not None:
              return cached_value
          
          # Вычисляем и кешируем
          result = func(*args, **kwargs)
          backend.set(cache_key, result, ttl)
          return result
      return wrapper
  return decorator
 
# Использование с любым бэкендом
memory = MemoryCache()
 
@cached(memory, ttl=60)
def expensive_calculation(n: int) -> int:
  # Тяжёлые вычисления
  return sum(i * i for i in range(n))
 
# Первый вызов вычисляет
result1 = expensive_calculation(10000)  # Медленно
 
# Второй берёт из кеша
result2 = expensive_calculation(10000)  # Мгновенно

from typing import Protocol
 
class Validator(Protocol):
  def validate(self, value: Any) -> bool:
      """Проверка корректности значения"""
      ...
  
  def error_message(self) -> str:
      """Сообщение об ошибке"""
      ...
 
class EmailValidator:
  def validate(self, value: Any) -> bool:
      import re
      if not isinstance(value, str):
          return False
      pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
      return bool(re.match(pattern, value))
  
  def error_message(self) -> str:
      return "Некорректный формат email"
 
class RangeValidator:
  def __init__(self, min_val: int, max_val: int):
      self.min = min_val
      self.max = max_val
  
  def validate(self, value: Any) -> bool:
      if not isinstance(value, (int, float)):
          return False
      return self.min <= value <= self.max
  
  def error_message(self) -> str:
      return f"Значение должно быть от {self.min} до {self.max}"
 
def validate_field(value: Any, validators: list[Validator]) -> tuple[bool, list[str]]:
  """Проверка значения набором валидаторов"""
  errors = []
  for validator in validators:
      if not validator.validate(value):
          errors.append(validator.error_message())
  return len(errors) == 0, errors
 
# Композиция валидаторов
age_validators = [RangeValidator(0, 150)]
email_validators = [EmailValidator()]
 
is_valid, errors = validate_field(25, age_validators)  # (True, [])
is_valid, errors = validate_field("bad-email", email_validators)  # (False, [...])

from typing import Protocol, Callable, TypeVar
 
T = TypeVar('T')
 
class ErrorHandler(Protocol[T]):
  def handle(self, error: Exception, operation: Callable[[], T]) -> T:
      """Обработка ошибки и возврат результата"""
      ...
 
class RetryHandler:
  def __init__(self, attempts: int = 3):
      self.attempts = attempts
  
  def handle(self, error: Exception, operation: Callable[[], T]) -> T:
      for attempt in range(self.attempts):
          try:
              return operation()
          except Exception as e:
              if attempt == self.attempts - 1:
                  raise e
              # Повторяем попытку
      raise error
 
class FallbackHandler:
  def __init__(self, default_value: T):
      self.default = default_value
  
  def handle(self, error: Exception, operation: Callable[[], T]) -> T:
      try:
          return operation()
      except Exception:
          return self.default
 
def execute_with_handler(operation: Callable[[], T], handler: ErrorHandler[T]) -> T:
  """Выполнение операции с заданным обработчиком ошибок"""
  try:
      return operation()
  except Exception as e:
      return handler.handle(e, operation)
 
# Разные стратегии для разных операций
retry = RetryHandler(attempts=5)
fallback = FallbackHandler(default_value=[])
 
# Критичная операция - ретраи
result = execute_with_handler(lambda: fetch_user_data(), retry)
 
# Некритичная - фолбек к пустому списку
items = execute_with_handler(lambda: load_recommendations(), fallback)

from typing import Protocol, runtime_checkable
 
@runtime_checkable
class DataProcessor(Protocol):
    def process(self, data: dict) -> list[str]:
        """Обработка данных"""
        ...
 
class BrokenProcessor:
    def process(self, data: dict, options: dict) -> dict:  # Другая сигнатура!
        """Реализация с дополнительным параметром"""
        return {"result": "broken"}
 
# isinstance говорит True - метод process есть
broken = BrokenProcessor()
print(isinstance(broken, DataProcessor))  # True, но это ложь!
 
# А вот тут упадёт
def use_processor(proc: DataProcessor):
    result = proc.process({"key": "value"})  # TypeError: не хватает аргумента options!
    return result

from typing import Protocol
 
class BaseService:
    def __init__(self):
        self.initialized = False
    
    def setup(self) -> None:
        """Инициализация сервиса"""
        self.initialized = True
 
class Loggable(Protocol):
    def log(self, message: str) -> None:
        """Логирование"""
        ...
 
class UserService(BaseService):
    def log(self, message: str) -> None:
        """Добавляем логирование"""
        print(f"[USER] {message}")
    
    def create_user(self, name: str) -> None:
        if not self.initialized:
            self.setup()
        self.log(f"Создаём пользователя {name}")

from typing import Protocol
 
class Stack(Protocol):
    def push(self, item: int) -> None:
        """Добавление элемента"""
        ...
    
    def pop(self) -> int:
        """Извлечение элемента"""
        ...
 
class BrokenStack:
    def __init__(self):
        self.items: list[int] = []
    
    def push(self, item: int) -> None:
        # Реализация неправильная - добавляем в начало вместо конца!
        self.items.insert(0, item)
    
    def pop(self) -> int:
        # Извлекаем с конца
        return self.items.pop() if self.items else 0
 
# Статический анализ доволен - методы есть, типы совпадают
def use_stack(s: Stack) -> None:
    s.push(1)
    s.push(2)
    s.push(3)
    result = s.pop()  # Ожидаем 3, получим 1!
    print(result)

from typing import AsyncIterator
import asyncio
 
class AsyncDataStream:
    def __init__(self, data: list[int]):
        self.data = data
        self.index = 0
    
    # Забыли сделать методы асинхронными!
    def __aiter__(self):
        return self
    
    def __anext__(self) -> int:  # Должно быть async def!
        if self.index >= len(self.data):
            raise StopAsyncIteration
        value = self.data[self.index]
        self.index += 1
        return value
 
async def process_stream(stream: AsyncIterator[int]) -> None:
    # Упадёт с TypeError - __anext__ должен возвращать awaitable
    async for item in stream:
        print(item)

from typing import Protocol, TypeVar, Generic
 
T_in = TypeVar('T_in', contravariant=True)
T_out = TypeVar('T_out', covariant=True)
 
class Transformer(Protocol[T_in, T_out]):
    def transform(self, input_data: T_in) -> T_out:
        """Преобразование входных данных в выходные"""
        ...
 
# Попробуй быстро понять, какие типы куда подставляются
class StringToIntTransformer:
    def transform(self, input_data: str) -> int:
        return len(input_data)
 
# Это Transformer[str, int] или наоборот? А может Transformer[int, str]?
# Контравариантность и ковариантность меняют правила игры

from typing import Protocol, Union
 
class Plugin(Protocol):
    def execute(self) -> None: ...
 
class Plugin2(Protocol):
    def execute(self) -> None: ...
    def get_version(self) -> str: ...
 
def run_plugin(plugin: Union[Plugin, Plugin2]) -> None:
    plugin.execute()
    # Проверяем наличие метода в рантайме - некрасиво
    if hasattr(plugin, 'get_version'):
        print(f"Версия: {plugin.get_version()}")

from typing import Protocol, runtime_checkable
 
@runtime_checkable
class Drawable(Protocol):
    def draw(self) -> None: ...
 
class Shape:
    pass
 
# Добавляем метод в рантайме
shape = Shape()
shape.draw = lambda: print("Рисуем фигуру")  # type: ignore
 
print(isinstance(shape, Drawable))  # True - метод есть
# Но mypy считает Shape несоответствующим Drawable

from typing import Protocol, Any
 
class DataPlugin(Protocol):
    """Контракт для плагинов обработки данных"""
    
    @property
    def name(self) -> str:
        """Уникальное имя плагина"""
        ...
    
    @property
    def version(self) -> str:
        """Версия плагина"""
        ...
    
    def initialize(self, config: dict[str, Any]) -> None:
        """Инициализация плагина с конфигурацией"""
        ...
    
    def process(self, data: bytes) -> dict[str, Any]:
        """Обработка входных данных"""
        ...
    
    def can_handle(self, data_type: str) -> bool:
        """Проверка поддержки типа данных"""
        ...

import importlib
import importlib.util
from pathlib import Path
from typing import runtime_checkable
 
@runtime_checkable
class DataPlugin(Protocol):
    # ... определение из примера выше
 
class PluginManager:
    """Менеджер загрузки и управления плагинами"""
    
    def __init__(self, plugins_dir: Path):
        self.plugins_dir = plugins_dir
        self.plugins: dict[str, DataPlugin] = {}
        self._load_errors: list[tuple[str, Exception]] = []
    
    def discover_plugins(self) -> None:
        """Поиск и загрузка всех плагинов из директории"""
        if not self.plugins_dir.exists():
            raise ValueError(f"Директория плагинов не существует: {self.plugins_dir}")
        
        # Ищем Python-модули
        for plugin_file in self.plugins_dir.glob("*.py"):
            if plugin_file.name.startswith("_"):
                continue  # Пропускаем служебные файлы
            
            try:
                self._load_plugin(plugin_file)
            except Exception as e:
                # Собираем ошибки, но не падаем
                self._load_errors.append((plugin_file.name, e))
    
    def _load_plugin(self, plugin_path: Path) -> None:
        """Загрузка одного плагина из файла"""
        module_name = plugin_path.stem
        
        # Динамический импорт модуля
        spec = importlib.util.spec_from_file_location(module_name, plugin_path)
        if spec is None or spec.loader is None:
            raise ImportError(f"Не удалось загрузить спецификацию для {plugin_path}")
        
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        
        # Ищем классы, соответствующие протоколу
        for attr_name in dir(module):
            if attr_name.startswith("_"):
                continue
            
            attr = getattr(module, attr_name)
            
            # Проверяем, что это класс и соответствует протоколу
            if not isinstance(attr, type):
                continue
            
            # Создаем экземпляр и проверяем через isinstance
            try:
                instance = attr()
                if isinstance(instance, DataPlugin):
                    plugin_name = instance.name
                    if plugin_name in self.plugins:
                        print(f"Предупреждение: плагин {plugin_name} уже загружен, пропускаем")
                        continue
                    
                    self.plugins[plugin_name] = instance
                    print(f"Загружен плагин: {plugin_name} v{instance.version}")
            except Exception as e:
                # Не смогли создать экземпляр - пропускаем
                print(f"Ошибка при инстанцировании {attr_name}: {e}")
    
    def get_plugin_for_type(self, data_type: str) -> DataPlugin | None:
        """Поиск плагина, поддерживающего указанный тип данных"""
        for plugin in self.plugins.values():
            if plugin.can_handle(data_type):
                return plugin
        return None
    
    def process_data(self, data: bytes, data_type: str, config: dict[str, Any] | None = None) -> dict[str, Any]:
        """Обработка данных через подходящий плагин"""
        plugin = self.get_plugin_for_type(data_type)
        if plugin is None:
            raise ValueError(f"Нет плагина для обработки типа: {data_type}")
        
        # Инициализируем с конфигурацией если нужно
        if config:
            plugin.initialize(config)
        
        return plugin.process(data)
    
    @property
    def load_errors(self) -> list[tuple[str, Exception]]:
        """Список ошибок при загрузке плагинов"""
        return self._load_errors

import json
from typing import Any
 
class JSONPlugin:
    """Плагин для обработки JSON-данных"""
    
    def __init__(self):
        self._config: dict[str, Any] = {}
    
    @property
    def name(self) -> str:
        return "json_processor"
    
    @property
    def version(self) -> str:
        return "1.0.0"
    
    def initialize(self, config: dict[str, Any]) -> None:
        """Сохраняем конфигурацию"""
        self._config = config
    
    def process(self, data: bytes) -> dict[str, Any]:
        """Парсинг JSON и добавление метаданных"""
        parsed = json.loads(data.decode('utf-8'))
        
        # Добавляем метаданные из конфигурации
        result = {
            "data": parsed,
            "plugin": self.name,
            "version": self.version
        }
        
        if "add_timestamp" in self._config and self._config["add_timestamp"]:
            import time
            result["processed_at"] = time.time()
        
        return result
    
    def can_handle(self, data_type: str) -> bool:
        """Поддерживаем JSON"""
        return data_type.lower() in ("json", "application/json")

import xml.etree.ElementTree as ET
from typing import Any
 
class XMLPlugin:
    """Плагин для обработки XML-данных"""
    
    @property
    def name(self) -> str:
        return "xml_processor"
    
    @property
    def version(self) -> str:
        return "1.0.0"
    
    def initialize(self, config: dict[str, Any]) -> None:
        """XML-плагину конфигурация не нужна"""
        pass
    
    def process(self, data: bytes) -> dict[str, Any]:
        """Парсинг XML в словарь"""
        try:
            root = ET.fromstring(data.decode('utf-8'))
            
            # Преобразуем XML в словарь
            result = {
                "root_tag": root.tag,
                "attributes": root.attrib,
                "children": []
            }
            
            for child in root:
                result["children"].append({
                    "tag": child.tag,
                    "text": child.text,
                    "attributes": child.attrib
                })
            
            return result
            
        except ET.ParseError as e:
            # Возвращаем информацию об ошибке
            return {
                "error": "XML parsing failed",
                "message": str(e),
                "plugin": self.name
            }
    
    def can_handle(self, data_type: str) -> bool:
        """Поддерживаем XML"""
        return data_type.lower() in ("xml", "application/xml", "text/xml")

from typing import TYPE_CHECKING
 
if TYPE_CHECKING:
    # Импорты только для проверки типов
    from json_plugin import JSONPlugin
    from xml_plugin import XMLPlugin
    
    # mypy проверит, что классы соответствуют протоколу
    json_plugin: DataPlugin = JSONPlugin()
    xml_plugin: DataPlugin = XMLPlugin()

from pathlib import Path
 
# Инициализация менеджера
manager = PluginManager(Path("./plugins"))
manager.discover_plugins()
 
# Проверка ошибок загрузки
if manager.load_errors:
    print("Ошибки при загрузке плагинов:")
    for filename, error in manager.load_errors:
        print(f"  {filename}: {error}")
 
# Обработка JSON-данных
json_data = b'{"name": "test", "value": 42}'
result = manager.process_data(
    json_data,
    "json",
    config={"add_timestamp": True}
)
print(f"JSON результат: {result}")
 
# Обработка XML-данных
xml_data = b'<root><item>test</item></root>'
result = manager.process_data(xml_data, "xml")
print(f"XML результат: {result}")

from typing import Protocol
 
class PrioritizedPlugin(Protocol):
  """Протокол с поддержкой приоритетов"""
  
  name: str
  version: str
  
  @property
  def priority(self) -> int:
      """Приоритет плагина (больше = выше)"""
      ...
  
  def initialize(self, config: dict[str, Any]) -> None: ...
  def process(self, data: bytes) -> dict[str, Any]: ...
  def can_handle(self, data_type: str) -> bool: ...

def get_plugin_for_type(self, data_type: str) -> DataPlugin | None:
  """Поиск плагина с наивысшим приоритетом"""
  candidates = [
      plugin for plugin in self.plugins.values()
      if plugin.can_handle(data_type)
  ]
  
  if not candidates:
      return None
  
  # Сортируем по приоритету, если он есть
  def get_priority(plugin: DataPlugin) -> int:
      return getattr(plugin, 'priority', 0)
  
  return max(candidates, key=get_priority)

class Logger(Protocol):
  def log(self, level: str, message: str) -> None: ...
 
class PluginContext:
  """Контекст выполнения для плагинов"""
  
  def __init__(self, logger: Logger):
      self.logger = logger
      self.shared_cache: dict[str, Any] = {}
 
class ContextAwarePlugin(Protocol):
  """Плагин с доступом к контексту"""
  
  name: str
  version: str
  
  def set_context(self, context: PluginContext) -> None:
      """Установка контекста перед использованием"""
      ...
  
  def process(self, data: bytes) -> dict[str, Any]: ...
  def can_handle(self, data_type: str) -> bool: ...

class AdvancedPluginManager:
  def __init__(self, plugins_dir: Path, logger: Logger):
      self.plugins_dir = plugins_dir
      self.context = PluginContext(logger)
      self.plugins: dict[str, ContextAwarePlugin] = {}
  
  def _load_plugin(self, plugin_path: Path) -> None:
      # ... загрузка модуля
      
      instance = attr()
      
      # Передаём контекст если плагин его поддерживает
      if hasattr(instance, 'set_context'):
          instance.set_context(self.context)
      
      self.plugins[instance.name] = instance

from functools import wraps
from typing import Callable, TypeVar
 
T = TypeVar('T')
 
def safe_plugin_call(fallback_value: T) -> Callable:
  """Декоратор для безопасного вызова методов плагина"""
  def decorator(func: Callable[..., T]) -> Callable[..., T]:
      @wraps(func)
      def wrapper(*args, **kwargs) -> T:
          try:
              return func(*args, **kwargs)
          except Exception as e:
              # Логируем ошибку
              print(f"Ошибка в плагине: {e}")
              return fallback_value
      return wrapper
  return decorator
 
class RobustPluginWrapper:
  """Обёртка плагина с защитой от ошибок"""
  
  def __init__(self, plugin: DataPlugin):
      self._plugin = plugin
  
  @property
  def name(self) -> str:
      return self._plugin.name
  
  @safe_plugin_call(fallback_value={})
  def process(self, data: bytes) -> dict[str, Any]:
      return self._plugin.process(data)

import pytest
 
class MockPluginManager:
  """Упрощённый менеджер для тестов"""
  
  def __init__(self):
      self.plugins: dict[str, DataPlugin] = {}
  
  def register(self, plugin: DataPlugin) -> None:
      self.plugins[plugin.name] = plugin
 
def test_json_plugin():
  manager = MockPluginManager()
  plugin = JSONPlugin()
  manager.register(plugin)
  
  # Проверяем обработку
  data = b'{"test": "value"}'
  result = plugin.process(data)
  
  assert "data" in result
  assert result["data"]["test"] == "value"
  assert result["plugin"] == "json_processor"
 
def test_plugin_protocol_compliance():
  """Проверка соответствия протоколу"""
  plugin = JSONPlugin()
  
  # Статический анализатор проверит эти присваивания
  p: DataPlugin = plugin
  
  assert callable(p.process)
  assert callable(p.can_handle)
  assert isinstance(p.name, str)