Исходный код EJIO.user.repository

## /user/repository.py

from threading import Lock
from typing import Callable, Any
from .model import User

__all__ = [
    'UserRepository',
]




class UserRepository:
    """
    Внутренний репозиторий управления коллекцией пользователей в ОЗУ.
    Обеспечивает потокобезопасные CRUD и SQL-подобные операции выборки.
    """
    __slots__ = ('_lock', '_storage')

    def __init__(self) -> None:
        self._lock: Lock = Lock()
        # Карта соответствия: {user_id: User_dict}
        self._storage: dict[str, dict[str, Any]] = {}



    def add(self, user: User) -> None:
        with self._lock:
            self._storage[user.id] = user.to_dict()




    def get_by_id(self, user_id: str) -> User | None:
        with self._lock:
            data = self._storage.get(user_id)
            return User.from_dict(data) if data else None




    def get_by_login(self, login: str) -> User | None:
        """Ищет пользователя по уникальному логину (регистронезависимо)."""
        with self._lock:
            target = login.strip().lower()
            for data in self._storage.values():
                if data["login"].lower() == target:
                    return User.from_dict(data)
            return None




    def get_all(self) -> list[User]:
        with self._lock:
            return [User.from_dict(data) for data in self._storage.values()]




    def get_all_raw(self) -> list[dict[str, Any]]:
        with self._lock:
            return [u for u in self._storage.values()]




    def filter(self, predicate: Callable[[User], bool]) -> list[User]:
        """
        Выполняет продвинутую фильтрацию пользователей на основе лямбда-предиката.
        Аналог SQL WHERE.
        """
        with self._lock:
            results = []
            for data in self._storage.values():
                user_obj = User.from_dict(data)
                if predicate(user_obj):
                    results.append(user_obj)
            return results




    def load_raw_list(self, raw_list: list[dict[str, Any]]) -> None:
        """Быстрая атомарная заливка сырых словарей из файла конфигурации."""
        with self._lock:
            for data in raw_list:
                self._storage[data["id"]] = data




    def remove(self, user_id: str) -> None:
        with self._lock:
            if user_id in self._storage:
                del self._storage[user_id]




    def clear(self) -> None:
        with self._lock:
            self._storage.clear()