User manager¶

async def authenticate(
    self,
    identifier: str,
    password: str,
    *,
    login_identifier: LoginIdentifier | None = None,
) -> UP | None:
    """Return the matching user when credentials are valid.

    After successful verification, if the stored hash is deprecated (e.g. bcrypt),
    the password hash is upgraded to the current algorithm (e.g. Argon2) in the DB.
    If the DB update fails, login still succeeds without upgrading the hash.

    Args:
        identifier: Email or username string, depending on ``login_identifier``.
        password: Plain-text password.
        login_identifier: Lookup mode; defaults to :attr:`login_identifier` on this manager.
    """
    mode = login_identifier if login_identifier is not None else self.login_identifier
    user = await self._user_lifecycle.authenticate(
        identifier,
        password,
        login_identifier=mode,
        dummy_hash=_get_dummy_hash(),
        logger=logger,
    )
    if user is None:
        logger.warning("User login failed", extra={"event": "login_failed"})
        return None

    logger.info("User login succeeded", extra={"event": "login", "user_id": str(user.id)})
    return user

async def create(
    self,
    user_create: msgspec.Struct | Mapping[str, Any],
    *,
    safe: bool = True,
    allow_privileged: bool = False,
) -> UP:
    """Create a new user, hashing the provided password before persistence.

    Returns:
        The newly created user.
    """
    user = await self._user_lifecycle.create(user_create, safe=safe, allow_privileged=allow_privileged)
    logger.info("User registered", extra={"event": "register", "user_id": str(user.id)})
    return user

async def delete(self, user_id: ID) -> None:
    """Delete a user permanently and run the post-delete hook."""
    await self._user_lifecycle.delete(user_id)

async def forgot_password(self, email: str) -> None:
    """Trigger the forgot-password flow without revealing whether a user exists."""
    await self._account_tokens.forgot_password(email, dummy_hash=_get_dummy_hash())

async def get(self, user_id: ID) -> UP | None:
    """Return a user by identifier.

    Returns:
        The matching user when one exists, otherwise ``None``.
    """
    return await self._user_lifecycle.get(user_id)

async def list_users(self, *, offset: int, limit: int) -> tuple[list[UP], int]:
    """Return paginated users and the total available count."""
    return await self._user_lifecycle.list_users(offset=offset, limit=limit)

async def on_after_delete(self, user: UP) -> None:  # noqa: PLR6301
    """Hook invoked after a user is deleted permanently."""
    del user

async def on_after_forgot_password(  # noqa: PLR6301
    self,
    user: UP | None,
    token: str | None,
) -> None:
    """Hook invoked after a forgot-password request is processed.

    SECURITY: When ``user`` is ``None``, the email did not match any account.
    To prevent user enumeration via timing, your implementation MUST perform
    equivalent I/O in both cases (e.g., always enqueue a background task,
    whether or not an email will actually be sent). Do NOT conditionally
    skip work based on whether ``user`` is ``None``.
    """
    del user
    del token

async def on_after_login(self, user: UP) -> None:  # noqa: PLR6301
    """Hook invoked after a user authenticates successfully."""
    del user

async def on_after_register(self, user: UP, token: str) -> None:  # noqa: PLR6301
    """Hook invoked after a new user is created."""
    del user
    del token

async def on_after_request_verify_token(self, user: UP, token: str) -> None:  # noqa: PLR6301
    """Hook invoked after a verification token is requested."""
    del user
    del token

async def on_after_reset_password(self, user: UP) -> None:  # noqa: PLR6301
    """Hook invoked after a password reset completes."""
    del user

async def on_after_update(self, user: UP, update_dict: dict[str, Any]) -> None:  # noqa: PLR6301
    """Hook invoked after a user is updated successfully."""
    del user
    del update_dict

async def on_after_verify(self, user: UP) -> None:  # noqa: PLR6301
    """Hook invoked after a user verifies their email."""
    del user

async def on_before_delete(self, user: UP) -> None:  # noqa: PLR6301
    """Hook invoked before a user is deleted. Raise to cancel deletion."""
    del user

async def read_totp_secret(self, secret: str | None) -> str | None:
    """Return a plain-text TOTP secret from storage.

    Returns:
        Plain-text secret, or ``None`` when 2FA is disabled.
    """
    return await self._totp_secrets.read_secret(secret, load_cryptography_fernet=_load_cryptography_fernet)

async def request_verify_token(self, email: str) -> None:
    """Generate a new verification token for an existing unverified user."""
    await self._account_tokens.request_verify_token(email)

def require_account_state(self, user: UP, *, require_verified: bool = False) -> None:  # noqa: PLR6301
    """Validate account-state policy for authenticated flows.

    Args:
        user: User to validate.
        require_verified: When ``True``, also enforce ``is_verified``.
    """
    UserPolicy.require_account_state(user, require_verified=require_verified)

async def reset_password(self, token: str, password: str) -> UP:
    """Reset a user's password using a signed reset token.

    Returns:
        The updated user instance.
    """
    return await self._account_tokens.reset_password(token, password)

async def set_totp_secret(self, user: UP, secret: str | None) -> UP:
    """Store or clear the TOTP secret directly, bypassing None-filtering.

    Args:
        user: The user whose TOTP secret should be updated.
        secret: New secret string, or ``None`` to disable 2FA.

    Returns:
        The updated user instance.
    """
    return await self._totp_secrets.set_secret(
        user,
        secret,
        load_cryptography_fernet=_load_cryptography_fernet,
    )

async def update(self, user_update: msgspec.Struct | Mapping[str, Any], user: UP) -> UP:
    """Update mutable user fields, hashing passwords when provided.

    Fields with ``None`` values in *user_update* are treated as absent and
    will **not** overwrite existing data.  To explicitly clear a nullable
    field, use a dedicated method (e.g. ``set_totp_secret(user, None)``).

    Returns:
        The updated user, or the original user when there are no changes.
    """
    return await self._user_lifecycle.update(user_update, user)

async def verify(self, token: str) -> UP:
    """Mark a user as verified using a signed verification token.

    Returns:
        The verified user instance.
    """
    return await self._account_tokens.verify(token)

def write_verify_token(self, user: UP) -> str:
    """Return a signed email-verification token for a user.

    Returns:
        A short-lived verification token.
    """
    return self._account_tokens.write_verify_token(user)