xiaoxia-saas/packages/domain/auth/session_store.py

"""Session 存储端口（领域层接口）.

定义 Session 存储服务的抽象接口，具体实现由基础设施层（adapters）提供。
遵循 DDD 依赖倒置原则：领域层定义端口，外层实现端口。
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Optional


class SessionStorePort(ABC):
    """Session 存储端口（抽象接口）"""

    @abstractmethod
    def save_session(
        self,
        session_id: str,
        user_id: str,
        refresh_token: str,
        user_agent: str = "",
        ip_address: str = "",
        expires_in_days: int = 7,
    ) -> bool:
        """保存 Session"""
        ...

    @abstractmethod
    def get_session(self, session_id: str) -> Optional[dict]:
        """获取 Session"""
        ...

    @abstractmethod
    def get_session_by_refresh_token(self, refresh_token: str) -> Optional[dict]:
        """通过 refresh_token 查找 Session"""
        ...

    @abstractmethod
    def get_refresh_token(self, session_id: str) -> Optional[str]:
        """获取 refresh_token"""
        ...

    @abstractmethod
    def update_last_active(self, session_id: str) -> bool:
        """更新最后活跃时间"""
        ...

    @abstractmethod
    def delete_session(self, session_id: str) -> bool:
        """删除 Session"""
        ...

    @abstractmethod
    def get_user_sessions(self, user_id: str) -> list[dict]:
        """获取用户所有 Session"""
        ...

    @abstractmethod
    def delete_all_user_sessions(self, user_id: str) -> int:
        """删除用户所有 Session"""
        ...

    @abstractmethod
    def session_exists(self, session_id: str) -> bool:
        """检查 Session 是否存在"""
        ...