dzentra_bot/app/src/storage/quote_store.py

# app/src/storage/quote_store.py

from __future__ import annotations

from typing import Protocol, runtime_checkable

from src.market_data.acquisition.models.quote import Quote
from src.storage.exceptions import QuoteStoreError


# Контракт runtime-хранилища канонических котировок.
@runtime_checkable
class QuoteStoreProtocol(Protocol):
    def get(
        self,
        source_name: str,
        symbol: str,
        *,
        runtime_key: str = "default",
    ) -> Quote | None:
        """Вернуть котировку или None, если запись отсутствует."""

    def set(
        self,
        source_name: str,
        quote: Quote,
        *,
        runtime_key: str = "default",
    ) -> None:
        """Сохранить каноническую котировку без копирования модели."""

    def clear(
        self,
        source_name: str | None = None,
        symbol: str | None = None,
        *,
        runtime_key: str | None = None,
    ) -> None:
        """Удалить записи, соответствующие переданным фильтрам."""


# In-memory реализация runtime-хранилища канонических котировок.
class InMemoryQuoteStore:
    def __init__(self) -> None:
        self._items: dict[tuple[str, str, str], Quote] = {}

    def get(
        self,
        source_name: str,
        symbol: str,
        *,
        runtime_key: str = "default",
    ) -> Quote | None:
        return self._items.get(
            self._key(
                source_name=source_name,
                symbol=symbol,
                runtime_key=runtime_key,
            )
        )

    def set(
        self,
        source_name: str,
        quote: Quote,
        *,
        runtime_key: str = "default",
    ) -> None:
        normalized_source_name = self._normalize_source_name(
            source_name
        )
        normalized_runtime_key = self._normalize_runtime_key(
            runtime_key
        )

        if not isinstance(quote, Quote):
            raise QuoteStoreError(
                "Quote Store принимает только объект Quote."
            )

        normalized_symbol = self._normalize_symbol(
            quote.symbol
        )

        self._items[
            (
                normalized_source_name,
                normalized_runtime_key,
                normalized_symbol,
            )
        ] = quote

    def clear(
        self,
        source_name: str | None = None,
        symbol: str | None = None,
        *,
        runtime_key: str | None = None,
    ) -> None:
        if (
            source_name is None
            and symbol is None
            and runtime_key is None
        ):
            self._items.clear()
            return

        normalized_source_name = (
            self._normalize_source_name(source_name)
            if source_name is not None
            else None
        )
        normalized_symbol = (
            self._normalize_symbol(symbol)
            if symbol is not None
            else None
        )
        normalized_runtime_key = (
            self._normalize_runtime_key(runtime_key)
            if runtime_key is not None
            else None
        )

        keys_to_delete = [
            key
            for key in self._items
            if self._matches_filters(
                key,
                source_name=normalized_source_name,
                symbol=normalized_symbol,
                runtime_key=normalized_runtime_key,
            )
        ]

        for key in keys_to_delete:
            self._items.pop(key, None)

    def _key(
        self,
        *,
        source_name: str,
        symbol: str,
        runtime_key: str,
    ) -> tuple[str, str, str]:
        return (
            self._normalize_source_name(source_name),
            self._normalize_runtime_key(runtime_key),
            self._normalize_symbol(symbol),
        )

    def _matches_filters(
        self,
        key: tuple[str, str, str],
        *,
        source_name: str | None,
        symbol: str | None,
        runtime_key: str | None,
    ) -> bool:
        key_source_name, key_runtime_key, key_symbol = key

        if (
            source_name is not None
            and key_source_name != source_name
        ):
            return False

        if (
            runtime_key is not None
            and key_runtime_key != runtime_key
        ):
            return False

        if symbol is not None and key_symbol != symbol:
            return False

        return True

    def _normalize_source_name(
        self,
        source_name: str,
    ) -> str:
        normalized_source_name = str(source_name or "").strip()

        if not normalized_source_name:
            raise QuoteStoreError(
                "Имя источника Quote Store не должно быть пустым."
            )

        return normalized_source_name

    def _normalize_runtime_key(
        self,
        runtime_key: str,
    ) -> str:
        normalized_runtime_key = str(runtime_key or "").strip().lower()

        if not normalized_runtime_key:
            raise QuoteStoreError(
                "Runtime key Quote Store не должен быть пустым."
            )

        return normalized_runtime_key

    def _normalize_symbol(
        self,
        symbol: str,
    ) -> str:
        normalized_symbol = str(symbol or "").strip().upper()

        if not normalized_symbol:
            raise QuoteStoreError(
                "Символ Quote Store не должен быть пустым."
            )

        return normalized_symbol



QUOTE_RUNTIME_SOURCE_NAME = "legacy-market-price-cache"
_SHARED_QUOTE_STORE: QuoteStoreProtocol = InMemoryQuoteStore()


def get_quote_store() -> QuoteStoreProtocol:
    """Вернуть общий runtime-экземпляр канонического Quote Store."""
    return _SHARED_QUOTE_STORE