Email OTP
reauth.factors.email_otp
EmailOTPEnrollment
dataclass
EmailOTPEnrollment(
id: Any | None, identity_id: Any, email: str
)
id
instance-attribute
id: Any | None
identity_id
instance-attribute
identity_id: Any
email
instance-attribute
email: str
EmailOTP
dataclass
EmailOTP(
id: Any | None,
identity_id: Any | None,
email: str,
code_verifier: str,
expires_at: int,
authentication_session_id: Any,
)
id
instance-attribute
id: Any | None
identity_id
instance-attribute
identity_id: Any | None
email
instance-attribute
email: str
code_verifier
instance-attribute
code_verifier: str
expires_at
instance-attribute
expires_at: int
authentication_session_id
instance-attribute
authentication_session_id: Any
is_expired
is_expired() -> bool
Check if the OTP has expired.
Returns:
| Type | Description |
|---|---|
bool
|
True if the OTP has expired, False otherwise. |
EmailOTPException
EmailOTPException(message: str | None = None)
InvalidOTPException
InvalidOTPException(message: str | None = None)
ExpiredOTPException
ExpiredOTPException(message: str | None = None)
EmailOTPFactor
EmailOTPFactor(
*,
verifier: TokenVerifier,
code_length: int = 6,
lifetime: timedelta = datetime.timedelta(minutes=10),
identifier: str = "email_otp",
step: int = 0,
advance_by: int = 1,
)
Bases: FactorBase[EmailOTPEnrollment], ABC
AMR
class-attribute
AMR: AuthenticationMethodReference = (
AuthenticationMethodReference.EMAIL
)
verifier
instance-attribute
verifier = verifier
code_length
instance-attribute
code_length = code_length
lifetime
instance-attribute
lifetime = lifetime
create
async
create(
email: str,
authentication_session_id: Any,
identity_id: Any | None = None,
) -> tuple[str, EmailOTP]
Create a new OTP for the given identity.
If an existing OTP for the same authentication session exists, it's deleted and replaced with the new one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
email
|
str
|
The email address this OTP is sent to. |
required |
authentication_session_id
|
Any
|
The ID of the authentication session this OTP is associated with. |
required |
identity_id
|
Any | None
|
Optional ID of the identity to create the OTP for. Can be None for signup flows. |
None
|
Returns:
| Type | Description |
|---|---|
tuple[str, EmailOTP]
|
A tuple of (OTP code, EmailOTP instance). |
consume
async
consume(
code: str, authentication_session_id: Any
) -> tuple[typing.Any | None, str]
Consume an OTP code, deleting it from the persistent store if valid.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
str
|
The OTP code to consume. |
required |
authentication_session_id
|
Any
|
The ID of the authentication session this OTP is associated with. |
required |
Returns:
| Type | Description |
|---|---|
Any | None
|
A tuple of (identity_id, email). identity_id is None for new users (signup flow). |
str
|
For signup flows where identity_id is None, the application MUST create an |
tuple[Any | None, str]
|
identity and EmailOTPEnrollment after successful OTP verification. |
Raises:
| Type | Description |
|---|---|
InvalidOTPException
|
If the code is invalid or does not correspond to any OTP. |
ExpiredOTPException
|
If the OTP has expired. |
insert
abstractmethod
async
insert(email_otp: EmailOTP) -> typing.Any
Insert an EmailOTP instance into a persistent store.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
email_otp
|
EmailOTP
|
The EmailOTP instance to insert. |
required |
Returns:
| Type | Description |
|---|---|
Any
|
The ID of the inserted EmailOTP. |
Implementation contract
Persist the supplied OTP and return its ID. Enforce at most one pending OTP per authentication session. Leave commit and rollback to the caller.
get_by_authentication_session_id
abstractmethod
async
get_by_authentication_session_id(
authentication_session_id: Any,
) -> EmailOTP | None
Retrieve an EmailOTP instance by its authentication session from the persistent store.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
authentication_session_id
|
Any
|
The ID of the authentication session this OTP is associated with. |
required |
Returns:
| Type | Description |
|---|---|
EmailOTP | None
|
The corresponding EmailOTP instance, or None if not found. |
Implementation contract
Match the authentication session ID. Include expired records without validating them. Acquire protection before reading and retain it until the caller's transaction ends. Leave commit and rollback to the caller.
delete
abstractmethod
async
delete(email_otp: EmailOTP) -> None
Delete an EmailOTP instance from the persistent store.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
email_otp
|
EmailOTP
|
The EmailOTP instance to delete. |
required |
Implementation contract
Delete the supplied OTP by ID in the transaction that protected its lookup. Preserve that protection through consumption and leave commit and rollback to the caller.
delete_by_authentication_session_id
abstractmethod
async
delete_by_authentication_session_id(
authentication_session_id: Any,
) -> None
Delete all EmailOTP instances associated with a given authentication session ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
authentication_session_id
|
Any
|
The ID of the authentication session to delete OTPs for. |
required |
Implementation contract
Delete all pending OTPs for the supplied authentication session. Acquire protection for that session before deleting, including when no OTP exists, for example by locking a stable parent row. Leave commit and rollback to the caller.