feat(storage): add StorageManager read logic with Redis -> MySQL fallback

app/storage_manager.py

@@ -1,10 +1,17 @@
+import json
 import logging
 import time
 
 from sqlalchemy import text
+from sqlalchemy.orm import sessionmaker
+
+from app.models import MarketData, SymbolTimestamp
 
 logger = logging.getLogger(__name__)
 
+# Redis 缓存 TTL：30 天
+CACHE_TTL_SECONDS = 30 * 24 * 60 * 60
+
 
 class StorageManager:
     """存储管理器骨架，封装 Redis / MySQL 可用性检测与三级降级状态。"""
@@ -75,6 +82,134 @@ class StorageManager:
         else:
             logger.warning("MySQL 不可用")
 
+    def get_market_data(self, symbol, data_type, periods):
+        """
+        读取行情数据：Redis → MySQL → None（由调用方降级到 SQLite）。
+
+        Args:
+            symbol: 品种合约代码
+            data_type: 数据类型，如 futures
+            periods: 周期，如 5min（对应 Redis key 中的 period）
+
+        Returns:
+            dict 或 None
+        """
+        cache_key = f"market_data:{symbol}:{periods}"
+
+        # 1. 检查 Redis 缓存
+        if self.check_redis():
+            try:
+                raw = self.redis_client.get(cache_key)
+                if raw:
+                    return json.loads(raw)
+            except Exception as e:
+                logger.warning(f"Redis 读取失败 [{cache_key}]: {e}")
+
+        # 2. Redis 未命中，检查 MySQL 可用性
+        if not self.check_mysql():
+            return None
+
+        # 3. 从 MySQL 读取
+        try:
+            SessionLocal = sessionmaker(bind=self.mysql_engine)
+            with SessionLocal() as db:
+                record = (
+                    db.query(MarketData)
+                    .filter_by(symbol=symbol, data_type=data_type, period=periods)
+                    .first()
+                )
+                if not record:
+                    return None
+
+                result = {
+                    "current_price": record.current_price,
+                    "timestamp": (
+                        record.fetched_at.isoformat() if record.fetched_at else None
+                    ),
+                    "candles": (
+                        json.loads(record.candles_json) if record.candles_json else []
+                    ),
+                }
+
+                # 4. 回填 Redis（TTL 30 天）
+                if self.check_redis():
+                    try:
+                        self.redis_client.setex(
+                            cache_key,
+                            CACHE_TTL_SECONDS,
+                            json.dumps(result, ensure_ascii=False),
+                        )
+                    except Exception as e:
+                        logger.warning(f"Redis 回填失败 [{cache_key}]: {e}")
+
+                return result
+        except Exception as e:
+            logger.warning(f"MySQL 读取失败 [{cache_key}]: {e}")
+            return None
+
+    def get_symbol_timestamp(self, symbol, data_type):
+        """
+        读取合约时间戳：Redis → MySQL → None（由调用方降级到 SQLite）。
+
+        Args:
+            symbol: 品种合约代码
+            data_type: 数据类型，如 futures
+
+        Returns:
+            dict 或 None
+        """
+        cache_key = f"symbol_timestamps:{symbol}"
+
+        # 1. 检查 Redis 缓存
+        if self.check_redis():
+            try:
+                raw = self.redis_client.get(cache_key)
+                if raw:
+                    return json.loads(raw)
+            except Exception as e:
+                logger.warning(f"Redis 读取失败 [{cache_key}]: {e}")
+
+        # 2. Redis 未命中，检查 MySQL 可用性
+        if not self.check_mysql():
+            return None
+
+        # 3. 从 MySQL 读取
+        try:
+            SessionLocal = sessionmaker(bind=self.mysql_engine)
+            with SessionLocal() as db:
+                record = (
+                    db.query(SymbolTimestamp)
+                    .filter_by(symbol=symbol, data_type=data_type)
+                    .first()
+                )
+                if not record:
+                    return None
+
+                result = {
+                    "last_refresh_at": (
+                        record.last_refresh_at.isoformat()
+                        if record.last_refresh_at
+                        else None
+                    ),
+                    "refresh_count": record.refresh_count,
+                }
+
+                # 4. 回填 Redis（TTL 30 天）
+                if self.check_redis():
+                    try:
+                        self.redis_client.setex(
+                            cache_key,
+                            CACHE_TTL_SECONDS,
+                            json.dumps(result, ensure_ascii=False),
+                        )
+                    except Exception as e:
+                        logger.warning(f"Redis 回填失败 [{cache_key}]: {e}")
+
+                return result
+        except Exception as e:
+            logger.warning(f"MySQL 读取失败 [{cache_key}]: {e}")
+            return None
+
 
 _storage_manager = None
 

tests/test_storage_manager.py

@@ -0,0 +1,291 @@
+import json
+import logging
+import time
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.storage_manager import StorageManager, get_storage_manager
+
+
+class TestStorageManagerSkeleton:
+    """StorageManager 骨架与降级检测测试。"""
+
+    def test_init_holds_client_and_engine_references(self):
+        """__init__ 应持有 redis_client / mysql_engine 引用。"""
+        redis_client = MagicMock()
+        mysql_engine = MagicMock()
+
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        assert manager.redis_client is redis_client
+        assert manager.mysql_engine is mysql_engine
+
+    def test_init_sets_default_flags(self):
+        """初始化时应将可用性标志设为 False。"""
+        manager = StorageManager(redis_client=None, mysql_engine=None)
+
+        assert manager.redis_available is False
+        assert manager.mysql_available is False
+
+    def test_check_redis_returns_cached_result_within_interval(self):
+        """30 秒间隔内应返回缓存的 Redis 可用性结果，不重新探测。"""
+        redis_client = MagicMock()
+        redis_client.ping.side_effect = Exception("redis down")
+
+        manager = StorageManager(redis_client=redis_client, mysql_engine=None)
+        first = manager.check_redis()
+        redis_client.ping.side_effect = None
+        second = manager.check_redis()
+
+        assert first is False
+        assert second is False
+        assert redis_client.ping.call_count == 1
+
+    def test_check_redis_recovers_after_interval(self):
+        """超过检测间隔后，Redis 恢复应返回 True。"""
+        redis_client = MagicMock()
+        redis_client.ping.side_effect = Exception("redis down")
+
+        manager = StorageManager(
+            redis_client=redis_client, mysql_engine=None, check_interval=0.1
+        )
+        assert manager.check_redis() is False
+
+        redis_client.ping.side_effect = None
+        time.sleep(0.15)
+
+        assert manager.check_redis() is True
+
+    def test_check_redis_false_when_client_is_none(self):
+        """redis_client 为 None 时 check_redis 返回 False。"""
+        manager = StorageManager(redis_client=None, mysql_engine=None)
+
+        assert manager.check_redis() is False
+
+    def test_check_mysql_returns_cached_result_within_interval(self):
+        """30 秒间隔内应返回缓存的 MySQL 可用性结果，不重新探测。"""
+        mysql_engine = MagicMock()
+        mysql_engine.connect.side_effect = Exception("mysql down")
+
+        manager = StorageManager(redis_client=None, mysql_engine=mysql_engine)
+        first = manager.check_mysql()
+        mysql_engine.connect.side_effect = None
+        second = manager.check_mysql()
+
+        assert first is False
+        assert second is False
+        assert mysql_engine.connect.call_count == 1
+
+    def test_check_mysql_recovers_after_interval(self):
+        """超过检测间隔后，MySQL 恢复应返回 True。"""
+        mysql_engine = MagicMock()
+        mysql_engine.connect.side_effect = Exception("mysql down")
+
+        manager = StorageManager(
+            redis_client=None, mysql_engine=mysql_engine, check_interval=0.1
+        )
+        assert manager.check_mysql() is False
+
+        mysql_engine.connect.side_effect = None
+        time.sleep(0.15)
+
+        assert manager.check_mysql() is True
+
+    def test_check_mysql_false_when_engine_is_none(self):
+        """mysql_engine 为 None 时 check_mysql 返回 False。"""
+        manager = StorageManager(redis_client=None, mysql_engine=None)
+
+        assert manager.check_mysql() is False
+
+    def test_is_available_true_when_redis_available(self):
+        """Redis 可用时 is_available 返回 True。"""
+        redis_client = MagicMock()
+        manager = StorageManager(redis_client=redis_client, mysql_engine=None)
+
+        assert manager.is_available() is True
+
+    def test_is_available_true_when_mysql_available(self):
+        """MySQL 可用时 is_available 返回 True。"""
+        mysql_engine = MagicMock()
+        manager = StorageManager(redis_client=None, mysql_engine=mysql_engine)
+
+        assert manager.is_available() is True
+
+    def test_is_available_false_when_both_unavailable(self):
+        """Redis 与 MySQL 均不可用时 is_available 返回 False。"""
+        redis_client = MagicMock()
+        redis_client.ping.side_effect = Exception("redis down")
+        mysql_engine = MagicMock()
+        mysql_engine.connect.side_effect = Exception("mysql down")
+
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        assert manager.is_available() is False
+
+    def test_initialize_checks_backends_and_logs(self, caplog):
+        """initialize 应检测后端可用性并输出日志。"""
+        redis_client = MagicMock()
+        mysql_engine = MagicMock()
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        with caplog.at_level(logging.INFO):
+            manager.initialize()
+
+        assert "Redis 可用" in caplog.text
+        assert "MySQL 可用" in caplog.text
+
+    def test_initialize_logs_fallback_when_unavailable(self, caplog):
+        """后端不可用时 initialize 应输出降级日志。"""
+        manager = StorageManager(redis_client=None, mysql_engine=None)
+
+        with caplog.at_level(logging.WARNING):
+            manager.initialize()
+
+        assert "Redis 不可用" in caplog.text
+        assert "MySQL 不可用" in caplog.text
+
+
+class TestStorageManagerSingleton:
+    """全局单例工厂函数测试。"""
+
+    def test_get_storage_manager_returns_same_instance(self):
+        """多次调用 get_storage_manager 应返回同一实例。"""
+        first = get_storage_manager()
+        second = get_storage_manager()
+
+        assert first is second
+
+    def test_get_storage_manager_returns_storage_manager(self):
+        """get_storage_manager 应返回 StorageManager 实例。"""
+        manager = get_storage_manager()
+
+        assert isinstance(manager, StorageManager)
+
+
+class TestStorageManagerReadLogic:
+    """StorageManager Redis → MySQL 读取逻辑测试。"""
+
+    def test_get_market_data_returns_redis_value_on_cache_hit(self):
+        """Redis 命中时直接返回反序列化数据，不访问 MySQL。"""
+        redis_client = MagicMock()
+        mysql_engine = MagicMock()
+        expected = {
+            "current_price": 123.45,
+            "timestamp": "2026-07-04T10:00:00",
+            "candles": [{"open": 100, "close": 110}],
+        }
+        redis_client.get.return_value = json.dumps(expected)
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        result = manager.get_market_data("AG2606", "futures", "5min")
+
+        assert result == expected
+        redis_client.get.assert_called_once_with("market_data:AG2606:5min")
+        mysql_engine.connect.assert_not_called()
+
+    def test_get_market_data_falls_back_to_mysql_and_backfills_redis(self):
+        """Redis 未命中时从 MySQL 读取并回填 Redis（TTL 30 天）。"""
+        redis_client = MagicMock()
+        redis_client.get.return_value = None
+        mysql_engine = MagicMock()
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        fetched_at = datetime(2026, 7, 4, 10, 0, 0)
+        record = MagicMock()
+        record.current_price = 123.45
+        record.fetched_at = fetched_at
+        record.candles_json = json.dumps([{"open": 100, "close": 110}])
+
+        mock_query = MagicMock()
+        mock_query.filter_by.return_value.first.return_value = record
+        mock_session = MagicMock()
+        mock_session.return_value = mock_session
+        mock_session.query.return_value = mock_query
+        mock_session.__enter__ = MagicMock(return_value=mock_session)
+        mock_session.__exit__ = MagicMock(return_value=False)
+
+        mock_sessionmaker = MagicMock(return_value=mock_session)
+
+        with patch("app.storage_manager.sessionmaker", mock_sessionmaker):
+            result = manager.get_market_data("AG2606", "futures", "5min")
+
+        expected = {
+            "current_price": 123.45,
+            "timestamp": "2026-07-04T10:00:00",
+            "candles": [{"open": 100, "close": 110}],
+        }
+        assert result == expected
+        mock_sessionmaker.assert_called_once_with(bind=mysql_engine)
+        mock_query.filter_by.assert_called_once_with(
+            symbol="AG2606", data_type="futures", period="5min"
+        )
+        redis_client.setex.assert_called_once_with(
+            "market_data:AG2606:5min", 2592000, json.dumps(expected, ensure_ascii=False)
+        )
+
+    def test_get_market_data_returns_none_when_mysql_unavailable(self):
+        """Redis 和 MySQL 均不可用时返回 None，由调用方降级到 SQLite。"""
+        redis_client = MagicMock()
+        redis_client.ping.side_effect = Exception("redis down")
+        mysql_engine = MagicMock()
+        mysql_engine.connect.side_effect = Exception("mysql down")
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        result = manager.get_market_data("AG2606", "futures", "5min")
+
+        assert result is None
+
+    def test_get_symbol_timestamp_returns_redis_value_on_cache_hit(self):
+        """Redis 命中时直接返回合约时间戳数据，不访问 MySQL。"""
+        redis_client = MagicMock()
+        mysql_engine = MagicMock()
+        expected = {
+            "last_refresh_at": "2026-07-04T10:00:00",
+            "refresh_count": 42,
+        }
+        redis_client.get.return_value = json.dumps(expected)
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        result = manager.get_symbol_timestamp("AG2606", "futures")
+
+        assert result == expected
+        redis_client.get.assert_called_once_with("symbol_timestamps:AG2606")
+        mysql_engine.connect.assert_not_called()
+
+    def test_get_symbol_timestamp_falls_back_to_mysql_and_backfills_redis(self):
+        """Redis 未命中时从 MySQL 读取时间戳并回填 Redis（TTL 30 天）。"""
+        redis_client = MagicMock()
+        redis_client.get.return_value = None
+        mysql_engine = MagicMock()
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        last_refresh = datetime(2026, 7, 4, 10, 0, 0)
+        record = MagicMock()
+        record.last_refresh_at = last_refresh
+        record.refresh_count = 42
+
+        mock_query = MagicMock()
+        mock_query.filter_by.return_value.first.return_value = record
+        mock_session = MagicMock()
+        mock_session.return_value = mock_session
+        mock_session.query.return_value = mock_query
+        mock_session.__enter__ = MagicMock(return_value=mock_session)
+        mock_session.__exit__ = MagicMock(return_value=False)
+
+        mock_sessionmaker = MagicMock(return_value=mock_session)
+
+        with patch("app.storage_manager.sessionmaker", mock_sessionmaker):
+            result = manager.get_symbol_timestamp("AG2606", "futures")
+
+        expected = {
+            "last_refresh_at": "2026-07-04T10:00:00",
+            "refresh_count": 42,
+        }
+        assert result == expected
+        mock_sessionmaker.assert_called_once_with(bind=mysql_engine)
+        mock_query.filter_by.assert_called_once_with(symbol="AG2606", data_type="futures")
+        redis_client.setex.assert_called_once_with(
+            "symbol_timestamps:AG2606", 2592000, json.dumps(expected, ensure_ascii=False)
+        )