Модуль репозитория пользователей

Модуль repository реализует паттерн Repository для работы с данными пользователей. Содержит абстрактный базовый класс и его реализацию на SQLAlchemy.

ABCUsersRepository

Абстрактный базовый класс, определяющий интерфейс для работы с хранилищем пользователей.

Документация API

Bases: ABC

Абстрактный базовый класс для репозитория пользователей.

Определяет интерфейс для работы с хранилищем пользователей, который должен быть реализован в конкретных классах-наследниках.

Source code in src/adapters/repository.py

class ABCUsersRepository(abc.ABC):
    """Абстрактный базовый класс для репозитория пользователей.

    Определяет интерфейс для работы с хранилищем пользователей,
    который должен быть реализован в конкретных классах-наследниках.
    """

    @abc.abstractmethod
    async def update_login_time(self, user_id: uuid.UUID) -> None:
        """Обновить время последнего входа пользователя.

        Args:
            user_id: Уникальный идентификатор пользователя.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def activate(self, user_id: uuid.UUID) -> None:
        """Активировать пользователя.

        Args:
            user_id: Уникальный идентификатор пользователя.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def deactivate(self, user_id: uuid.UUID) -> None:
        """Деактивировать пользователя.

        Args:
            user_id: Уникальный идентификатор пользователя.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def verify_email(self, user_id: uuid.UUID) -> None:
        """Подтвердить email пользователя.

        Args:
            user_id: Уникальный идентификатор пользователя.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def verify_password(self, user_id: uuid.UUID, password: str) -> bool:
        """Проверяет корректность пароля пользователя.

        Args:
            user_id: Уникальный идентификатор пользователя.
            password: Пароль для проверки.

        Returns:
            bool: True, если пароль верный, иначе False.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def set_password(self, user_id: uuid.UUID, password: str) -> None:
        """Устанавливает новый пароль для пользователя.

        Args:
            user_id: Уникальный идентификатор пользователя.
            password: Новый пароль.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def add(self, user: User) -> None:
        """Добавляет пользователя в репозиторий.

        Args:
            user: Объект пользователя для добавления.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def get_by_email(self, email: str) -> User | None:
        """Находит пользователя по email.

        Args:
            email: Email пользователя для поиска.

        Returns:
            User | None: Объект пользователя, если найден, иначе None.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def get_by_id(self, user_id: uuid.UUID) -> User | None:
        """Находит пользователя по уникальному идентификатору.

        Args:
            user_id: UUID пользователя для поиска.

        Returns:
            User | None: Объект пользователя, если найден, иначе None.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def get_by_username(self, username: str) -> User | None:
        """Находит пользователя по username.

        Args:
            username: Username пользователя для поиска.

        Returns:
            User | None: Объект пользователя, если найден, иначе None.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def list_all(self, *, only_active: bool = False) -> list[User]:
        """Возвращает список всех пользователей.

        Args:
            only_active: Если True, возвращает только активных пользователей.

        Returns:
            list[User]: Список пользователей.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def update(self, user: User) -> None:
        """Обновляет данные пользователя в репозитории.

        Args:
            user: Объект пользователя с обновлёнными данными.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

    @abc.abstractmethod
    async def remove(self, user_id: uuid.UUID) -> None:
        """Удаляет пользователя из репозитория.

        Args:
            user_id: UUID пользователя для удаления.

        Raises:
            NotImplementedError: Если метод не переопределен в подклассе.

        """
        raise NotImplementedError

activate(user_id: uuid.UUID) -> None abstractmethod async

Активировать пользователя.

Parameters:

Name	Type	Description	Default
user_id	UUID	Уникальный идентификатор пользователя.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def activate(self, user_id: uuid.UUID) -> None:
    """Активировать пользователя.

    Args:
        user_id: Уникальный идентификатор пользователя.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

add(user: User) -> None abstractmethod async

Добавляет пользователя в репозиторий.

Parameters:

Name	Type	Description	Default
user	User	Объект пользователя для добавления.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def add(self, user: User) -> None:
    """Добавляет пользователя в репозиторий.

    Args:
        user: Объект пользователя для добавления.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

deactivate(user_id: uuid.UUID) -> None abstractmethod async

Деактивировать пользователя.

Parameters:

Name	Type	Description	Default
user_id	UUID	Уникальный идентификатор пользователя.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def deactivate(self, user_id: uuid.UUID) -> None:
    """Деактивировать пользователя.

    Args:
        user_id: Уникальный идентификатор пользователя.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

get_by_email(email: str) -> User | None abstractmethod async

Находит пользователя по email.

Parameters:

Name	Type	Description	Default
email	str	Email пользователя для поиска.	required

Returns:

Type	Description
User | None	User | None: Объект пользователя, если найден, иначе None.

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def get_by_email(self, email: str) -> User | None:
    """Находит пользователя по email.

    Args:
        email: Email пользователя для поиска.

    Returns:
        User | None: Объект пользователя, если найден, иначе None.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

get_by_id(user_id: uuid.UUID) -> User | None abstractmethod async

Находит пользователя по уникальному идентификатору.

Parameters:

Name	Type	Description	Default
user_id	UUID	UUID пользователя для поиска.	required

Returns:

Type	Description
User | None	User | None: Объект пользователя, если найден, иначе None.

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def get_by_id(self, user_id: uuid.UUID) -> User | None:
    """Находит пользователя по уникальному идентификатору.

    Args:
        user_id: UUID пользователя для поиска.

    Returns:
        User | None: Объект пользователя, если найден, иначе None.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

get_by_username(username: str) -> User | None abstractmethod async

Находит пользователя по username.

Parameters:

Name	Type	Description	Default
username	str	Username пользователя для поиска.	required

Returns:

Type	Description
User | None	User | None: Объект пользователя, если найден, иначе None.

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def get_by_username(self, username: str) -> User | None:
    """Находит пользователя по username.

    Args:
        username: Username пользователя для поиска.

    Returns:
        User | None: Объект пользователя, если найден, иначе None.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

list_all(*, only_active: bool = False) -> list[User] abstractmethod async

Возвращает список всех пользователей.

Parameters:

Name	Type	Description	Default
only_active	bool	Если True, возвращает только активных пользователей.	False

Returns:

Type	Description
list[User]	list[User]: Список пользователей.

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def list_all(self, *, only_active: bool = False) -> list[User]:
    """Возвращает список всех пользователей.

    Args:
        only_active: Если True, возвращает только активных пользователей.

    Returns:
        list[User]: Список пользователей.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

remove(user_id: uuid.UUID) -> None abstractmethod async

Удаляет пользователя из репозитория.

Parameters:

Name	Type	Description	Default
user_id	UUID	UUID пользователя для удаления.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def remove(self, user_id: uuid.UUID) -> None:
    """Удаляет пользователя из репозитория.

    Args:
        user_id: UUID пользователя для удаления.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

set_password(user_id: uuid.UUID, password: str) -> None abstractmethod async

Устанавливает новый пароль для пользователя.

Parameters:

Name	Type	Description	Default
user_id	UUID	Уникальный идентификатор пользователя.	required
password	str	Новый пароль.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def set_password(self, user_id: uuid.UUID, password: str) -> None:
    """Устанавливает новый пароль для пользователя.

    Args:
        user_id: Уникальный идентификатор пользователя.
        password: Новый пароль.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

update(user: User) -> None abstractmethod async

Обновляет данные пользователя в репозитории.

Parameters:

Name	Type	Description	Default
user	User	Объект пользователя с обновлёнными данными.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def update(self, user: User) -> None:
    """Обновляет данные пользователя в репозитории.

    Args:
        user: Объект пользователя с обновлёнными данными.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

update_login_time(user_id: uuid.UUID) -> None abstractmethod async

Обновить время последнего входа пользователя.

Parameters:

Name	Type	Description	Default
user_id	UUID	Уникальный идентификатор пользователя.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def update_login_time(self, user_id: uuid.UUID) -> None:
    """Обновить время последнего входа пользователя.

    Args:
        user_id: Уникальный идентификатор пользователя.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

verify_email(user_id: uuid.UUID) -> None abstractmethod async

Подтвердить email пользователя.

Parameters:

Name	Type	Description	Default
user_id	UUID	Уникальный идентификатор пользователя.	required

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def verify_email(self, user_id: uuid.UUID) -> None:
    """Подтвердить email пользователя.

    Args:
        user_id: Уникальный идентификатор пользователя.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

verify_password(user_id: uuid.UUID, password: str) -> bool abstractmethod async

Проверяет корректность пароля пользователя.

Parameters:

Name	Type	Description	Default
user_id	UUID	Уникальный идентификатор пользователя.	required
password	str	Пароль для проверки.	required

Returns:

Name	Type	Description
bool	bool	True, если пароль верный, иначе False.

Raises:

Type	Description
NotImplementedError	Если метод не переопределен в подклассе.

Source code in src/adapters/repository.py

@abc.abstractmethod
async def verify_password(self, user_id: uuid.UUID, password: str) -> bool:
    """Проверяет корректность пароля пользователя.

    Args:
        user_id: Уникальный идентификатор пользователя.
        password: Пароль для проверки.

    Returns:
        bool: True, если пароль верный, иначе False.

    Raises:
        NotImplementedError: Если метод не переопределен в подклассе.

    """
    raise NotImplementedError

SQLAlchemyUsersRepository

Реализация репозитория пользователей на базе SQLAlchemy с асинхронной поддержкой.

Особенности

• Полностью асинхронная работа с базой данных
• Поддержка Unit of Work (UoW)
• Автоматическое преобразование между ORM-моделями и доменными объектами
• Встроенная валидация данных

Документация API

Bases: ABCUsersRepository

Реализация репозитория пользователей на SQLAlchemy (PostgreSQL, async) с поддержкой UoW.

Source code in src/adapters/repository.py

class SQLAlchemyUsersRepository(ABCUsersRepository):
    """Реализация репозитория пользователей на SQLAlchemy (PostgreSQL, async) с поддержкой UoW."""

    def __init__(self, session: AsyncSession, hasher: IPasswordHasher):
        """Инициализирует репозиторий пользователей.

        Args:
            session: Асинхронная сессия SQLAlchemy.
            hasher: Сервис для хеширования паролей.

        """
        self.session = session
        self._hasher = hasher

    async def add(self, user: User) -> None:
        stmt = users.insert().values(
            id=user.id,
            first_name=user.first_name,
            last_name=user.last_name,
            email=user.email,
            username=user.username,
            hashed_password=user.hashed_password,
            is_active=user.is_active,
            is_verified=user.is_verified,
            created_at=user.created_at,
            updated_at=user.updated_at,
            last_login_at=user.last_login_at,
        )
        await self.session.execute(stmt)

    async def get_by_email(self, email: str) -> User | None:
        result = await self.session.execute(select(users).where(users.c.email == email))
        row = result.first()
        return self._row_to_user(row) if row else None

    async def get_by_id(self, user_id: uuid.UUID) -> User | None:
        result = await self.session.execute(select(users).where(users.c.id == user_id))
        row = result.first()
        return self._row_to_user(row) if row else None

    async def get_by_username(self, username: str) -> User | None:
        result = await self.session.execute(select(users).where(users.c.username == username))
        row = result.first()
        return self._row_to_user(row) if row else None

    async def list_all(self, *, only_active: bool = False) -> list[User]:
        stmt = select(users)
        if only_active:
            stmt = stmt.where(users.c.is_active.is_(True))
        result = await self.session.execute(stmt)
        return [self._row_to_user(row) for row in result.fetchall()]

    async def update(self, user: User) -> None:
        stmt = (
            update(users)
            .where(users.c.id == user.id)
            .values(
                first_name=user.first_name,
                last_name=user.last_name,
                email=user.email,
                username=user.username,
                hashed_password=user.hashed_password,
                is_active=user.is_active,
                is_verified=user.is_verified,
                updated_at=datetime.datetime.now(datetime.UTC),
                last_login_at=user.last_login_at,
            )
        )
        await self.session.execute(stmt)

    async def remove(self, user_id: uuid.UUID) -> None:
        stmt = delete(users).where(users.c.id == user_id)
        await self.session.execute(stmt)

    async def update_login_time(self, user_id: uuid.UUID) -> None:
        stmt = (
            update(users)
            .where(users.c.id == user_id)
            .values(
                last_login_at=datetime.datetime.now(datetime.UTC),
                updated_at=datetime.datetime.now(datetime.UTC),
            )
        )
        await self.session.execute(stmt)

    async def activate(self, user_id: uuid.UUID) -> None:
        stmt = (
            update(users)
            .where(users.c.id == user_id)
            .values(
                is_active=True,
                updated_at=datetime.datetime.now(datetime.UTC),
            )
        )
        await self.session.execute(stmt)

    async def deactivate(self, user_id: uuid.UUID) -> None:
        stmt = (
            update(users)
            .where(users.c.id == user_id)
            .values(
                is_active=False,
                updated_at=datetime.datetime.now(datetime.UTC),
            )
        )
        await self.session.execute(stmt)

    async def verify_email(self, user_id: uuid.UUID) -> None:
        stmt = (
            update(users)
            .where(users.c.id == user_id)
            .values(
                is_verified=True,
                updated_at=datetime.datetime.now(datetime.UTC),
            )
        )
        await self.session.execute(stmt)

    async def verify_password(self, user_id: uuid.UUID, password: str) -> bool:
        user = await self.get_by_id(user_id)
        if not user:
            return False
        return self._hasher.verify_password(password, user.hashed_password)

    async def set_password(self, user_id: uuid.UUID, password: str) -> None:
        hashed = self._hasher.hash_password(password)
        stmt = (
            update(users)
            .where(users.c.id == user_id)
            .values(
                hashed_password=hashed,
                updated_at=datetime.datetime.now(datetime.UTC),
            )
        )
        await self.session.execute(stmt)

    def _row_to_user(self, row) -> User:
        """Конвертирует строку результата SQLAlchemy в доменную модель User.

        Args:
            row: Строка результата запроса SQLAlchemy.

        Returns:
            Экземпляр доменной модели User.

        Note:
            Внутренний метод, не предназначен для использования извне.

        """
        r = row[0] if isinstance(row, tuple) else row
        return User(
            user_id=r.id,
            first_name=r.first_name,
            last_name=r.last_name,
            email=r.email,
            username=r.username,
            hashed_password=r.hashed_password,
            _hasher=self._hasher,
            is_active=r.is_active,
            is_verified=r.is_verified,
            created_at=r.created_at,
            updated_at=r.updated_at,
            last_login_at=r.last_login_at,
        )

__init__(session: AsyncSession, hasher: IPasswordHasher)

Инициализирует репозиторий пользователей.

Parameters:

Name	Type	Description	Default
session	AsyncSession	Асинхронная сессия SQLAlchemy.	required
hasher	IPasswordHasher	Сервис для хеширования паролей.	required

Source code in src/adapters/repository.py

def __init__(self, session: AsyncSession, hasher: IPasswordHasher):
    """Инициализирует репозиторий пользователей.

    Args:
        session: Асинхронная сессия SQLAlchemy.
        hasher: Сервис для хеширования паролей.

    """
    self.session = session
    self._hasher = hasher

Примеры использования

Создание репозитория

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from src.adapters.repository import SQLAlchemyUsersRepository
from src.adapters.hasher import BcryptPasswordHasher

# Инициализация зависимостей
engine = create_async_engine("postgresql+asyncpg://user:password@localhost/db")
hasher = BcryptPasswordHasher()

# Создание сессии и репозитория
async with AsyncSession(engine) as session:
    repo = SQLAlchemyUsersRepository(session, hasher)

    # Использование репозитория
    user = await repo.get_by_email("user@example.com")

Работа с пользователями

# Добавление пользователя
new_user = User(
    email="new@example.com",
    hashed_password="secure_password",
    is_active=True
)
await repo.add(new_user)

# Поиск пользователя
user = await repo.get_by_email("new@example.com")

# Обновление данных
user.first_name = "Иван"
await repo.update(user)

# Удаление пользователя
await repo.remove(user.id)

Рекомендации

1. Управление сессиями
2. Всегда используйте контекстный менеджер async with для работы с сессиями

3. Не храните ссылки на сессии дольше необходимого

4. Обработка ошибок

5. Обрабатывайте специфические исключения SQLAlchemy

6. Используйте транзакции для группировки операций

7. Производительность

8. Используйте selectinload для загрузки связанных сущностей

9. Применяйте пагинацию для больших выборок

10. Тестирование

11. Используйте тестовую базу данных
12. Применяйте фикстуры для подготовки тестовых данных
13. Проверяйте изоляцию тестов

Была ли статья полезна?

🙂 🙁

⬅ ORM Vault ➡