long_articles_agentic_supply/src/infra/database/ports.py

"""存储后端抽象接口 (Ports) —— 定义所有存储引擎的统一契约

Domain 层只依赖这里的抽象，Infra 层提供具体实现。
"""

import logging
from abc import ABC, abstractmethod
from enum import Enum
from typing import Any, Optional

logger = logging.getLogger(__name__)


class BackendType(str, Enum):
    """存储后端类型枚举"""

    MYSQL = "mysql"
    POSTGRESQL = "postgresql"
    MILVUS = "milvus"
    REDIS = "redis"


class BaseBackend(ABC):
    """所有存储后端的抽象基类"""

    def __init__(self, name: str, config: Any):
        self.name = name
        self.config = config
        self._initialized = False

    @abstractmethod
    async def init(self) -> None:
        """初始化连接池或客户端，失败抛异常"""
        ...

    @abstractmethod
    async def close(self) -> None:
        """关闭连接池或客户端，安全可幂等"""
        ...

    @abstractmethod
    async def health(self) -> bool:
        """健康检查：ping 成功后返回 True"""
        ...

    @property
    def initialized(self) -> bool:
        return self._initialized


class SqlBackend(BaseBackend):
    """SQL 类后端的公共抽象（MySQL / PostgreSQL）"""

    @abstractmethod
    async def fetch(
        self,
        query: str,
        params: Optional[tuple] = None,
    ) -> list[dict]:
        """SELECT 多行"""
        ...

    @abstractmethod
    async def fetch_one(
        self,
        query: str,
        params: Optional[tuple] = None,
    ) -> Optional[dict]:
        """SELECT 单行"""
        ...

    @abstractmethod
    async def execute(
        self,
        query: str,
        params: Any = None,
        *,
        batch: bool = False,
    ) -> int:
        """INSERT/UPDATE/DELETE，返回受影响行数"""
        ...


class VectorBackend(BaseBackend):
    """向量数据库的公共抽象（Milvus 等）"""

    @abstractmethod
    async def insert(
        self,
        collection: str,
        vectors: list[list[float]],
        metadata: Optional[list[dict]] = None,
    ) -> list[int]:
        """插入向量，返回主键 ID 列表"""
        ...

    @abstractmethod
    async def search(
        self,
        collection: str,
        query_vectors: list[list[float]],
        top_k: int = 10,
        filter_expr: Optional[str] = None,
    ) -> list[list[dict]]:
        """向量相似搜索"""
        ...

    @abstractmethod
    async def delete(
        self,
        collection: str,
        ids: list[int],
    ) -> int:
        """按主键删除向量"""
        ...


class KvBackend(BaseBackend):
    """KV 存储的公共抽象（Redis 等）"""

    @abstractmethod
    async def get(self, key: str) -> Optional[str]:
        """获取键值"""
        ...

    @abstractmethod
    async def set(
        self,
        key: str,
        value: str,
        ttl: Optional[int] = None,
    ) -> bool:
        """写入键值，可选 TTL（秒）"""
        ...

    @abstractmethod
    async def delete(self, *keys: str) -> int:
        """删除键，返回已删除数量"""
        ...

    @abstractmethod
    async def exists(self, *keys: str) -> int:
        """检查键是否存在，返回存在的数量"""
        ...