xiaoxia-saas/packages/adapters/in_memory/user_repository.py

"""
用户仓储 In-Memory 实现
"""

from typing import Dict, Optional

from packages.domain.entities import User
from packages.ports.user_repository import UserRepository


class InMemoryUserRepository(UserRepository):
    """用户仓储内存实现"""

    def __init__(self):
        self._users: Dict[str, User] = {}
        self._email_index: Dict[str, str] = {}  # email -> user_id
        self._username_index: Dict[str, str] = {}  # username -> user_id
        self._verification_token_index: Dict[str, str] = {}  # token -> user_id
        self._reset_token_index: Dict[str, str] = {}  # token -> user_id
        self._wechat_openid_index: Dict[str, str] = {}  # openid -> user_id
        self._wechat_unionid_index: Dict[str, str] = {}  # unionid -> user_id
        self._phone_index: Dict[str, str] = {}  # phone -> user_id

    def save(self, user: User) -> None:
        """保存用户"""
        self._users[user.id] = user
        self._email_index[user.email.lower()] = user.id
        if user.username:
            self._username_index[user.username.lower()] = user.id
        if user.email_verification_token:
            self._verification_token_index[user.email_verification_token] = user.id
        if user.password_reset_token:
            self._reset_token_index[user.password_reset_token] = user.id
        if user.wechat_openid:
            self._wechat_openid_index[user.wechat_openid] = user.id
        if user.wechat_unionid:
            self._wechat_unionid_index[user.wechat_unionid] = user.id
        if user.phone:
            self._phone_index[user.phone] = user.id

    def find_by_id(self, user_id: str) -> Optional[User]:
        """根据 ID 查找用户"""
        return self._users.get(user_id)

    def find_by_email(self, email: str) -> Optional[User]:
        """根据邮箱查找用户"""
        user_id = self._email_index.get(email.lower())
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_username(self, username: str) -> Optional[User]:
        """根据用户名查找用户"""
        user_id = self._username_index.get(username.lower())
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_verification_token(self, token: str) -> Optional[User]:
        """根据邮箱验证令牌查找用户"""
        user_id = self._verification_token_index.get(token)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_password_reset_token(self, token: str) -> Optional[User]:
        """根据密码重置令牌查找用户"""
        user_id = self._reset_token_index.get(token)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_wechat_openid(self, openid: str) -> Optional[User]:
        """根据微信 openid 查找用户"""
        user_id = self._wechat_openid_index.get(openid)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_wechat_unionid(self, unionid: str) -> Optional[User]:
        """根据微信 unionid 查找用户"""
        if not unionid:
            return None
        user_id = self._wechat_unionid_index.get(unionid)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_phone(self, phone: str) -> Optional[User]:
        """根据手机号查找用户"""
        if not phone:
            return None
        user_id = self._phone_index.get(phone)
        if user_id:
            return self._users.get(user_id)
        return None

    def delete(self, user_id: str) -> bool:
        """删除用户"""
        user = self._users.get(user_id)
        if not user:
            return False

        # 清理索引
        self._email_index.pop(user.email.lower(), None)
        if user.username:
            self._username_index.pop(user.username.lower(), None)
        if user.email_verification_token:
            self._verification_token_index.pop(user.email_verification_token, None)
        if user.password_reset_token:
            self._reset_token_index.pop(user.password_reset_token, None)

        # 删除用户
        del self._users[user_id]
        return True