# This file is auto-generated by oagen. Do not edit.

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Union

if TYPE_CHECKING:
    from .._client import AsyncWorkOSClient, WorkOSClient

from .._types import RequestOptions, enum_value
from .models import (
    AuthenticationChallengeVerifyResponse,
    UserAuthenticationFactorEnrollResponse,
)
from workos.common.models.authentication_challenge import AuthenticationChallenge
from workos.common.models.authentication_factor import AuthenticationFactor
from workos.common.models.authentication_factor_enrolled import (
    AuthenticationFactorEnrolled,
)
from workos.common.models.authentication_factors_create_request_type import (
    AuthenticationFactorsCreateRequestType,
)
from workos.common.models.pagination_order import PaginationOrder
from .._pagination import AsyncPage, SyncPage


class MultiFactorAuth:
    """Multi Factor Auth API resources."""

    def __init__(self, client: "WorkOSClient") -> None:
        self._client = client

    def verify_challenge(
        self,
        id: str,
        *,
        code: str,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationChallengeVerifyResponse:
        """Verify Challenge

        Verifies an Authentication Challenge.

        Args:
            id: The unique ID of the Authentication Challenge.
            code: The one-time code to verify.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationChallengeVerifyResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "code": code,
        }
        return self._client.request(
            method="post",
            path=("auth", "challenges", str(id), "verify"),
            body=body,
            model=AuthenticationChallengeVerifyResponse,
            request_options=request_options,
        )

    def enroll_factor(
        self,
        *,
        type: Union[AuthenticationFactorsCreateRequestType, str],
        phone_number: Optional[str] = None,
        totp_issuer: Optional[str] = None,
        totp_user: Optional[str] = None,
        user_id: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationFactorEnrolled:
        """Enroll Factor

        Enrolls an Authentication Factor to be used as an additional factor of authentication. The returned ID should be used to create an authentication Challenge.

        Args:
            type: The type of factor to enroll.
            phone_number: Required when type is 'sms'.
            totp_issuer: Required when type is 'totp'.
            totp_user: Required when type is 'totp'.
            user_id: The ID of the user to associate the factor with.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationFactorEnrolled

        Raises:
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "type": enum_value(type),
                "phone_number": phone_number,
                "totp_issuer": totp_issuer,
                "totp_user": totp_user,
                "user_id": user_id,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="post",
            path=("auth", "factors", "enroll"),
            body=body,
            model=AuthenticationFactorEnrolled,
            request_options=request_options,
        )

    def get_factor(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationFactor:
        """Get Factor

        Gets an Authentication Factor.

        Args:
            id: The unique ID of the Factor.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationFactor

        Raises:
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return self._client.request(
            method="get",
            path=("auth", "factors", str(id)),
            model=AuthenticationFactor,
            request_options=request_options,
        )

    def delete_factor(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> None:
        """Delete Factor

        Permanently deletes an Authentication Factor. It cannot be undone.

        Args:
            id: The unique ID of the Factor.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        self._client.request(
            method="delete",
            path=("auth", "factors", str(id)),
            request_options=request_options,
        )

    def challenge_factor(
        self,
        id: str,
        *,
        sms_template: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationChallenge:
        """Challenge Factor

        Creates a Challenge for an Authentication Factor.

        Args:
            id: The unique ID of the Authentication Factor to be challenged.
            sms_template: A custom template for the SMS message. Use the {{code}} placeholder to include the verification code.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationChallenge

        Raises:
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "sms_template": sms_template,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="post",
            path=("auth", "factors", str(id), "challenge"),
            body=body,
            model=AuthenticationChallenge,
            request_options=request_options,
        )

    def list_user_auth_factors(
        self,
        userland_user_id: str,
        *,
        limit: Optional[int] = None,
        before: Optional[str] = None,
        after: Optional[str] = None,
        order: Optional[Union[PaginationOrder, str]] = "desc",
        request_options: Optional[RequestOptions] = None,
    ) -> SyncPage[AuthenticationFactor]:
        """List authentication factors

        Lists the [authentication factors](https://workos.com/docs/reference/authkit/mfa/authentication-factor) for a user.

        Args:
            userland_user_id: The ID of the [user](https://workos.com/docs/reference/authkit/user).
            limit: Upper limit on the number of objects to return, between `1` and `100`. Defaults to `10`.
            before: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list.
            after: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list.
            order: Order the results by the creation time. Defaults to `desc`.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            SyncPage[AuthenticationFactor]

        Raises:
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
            }.items()
            if v is not None
        }
        return self._client.request_page(
            method="get",
            path=("user_management", "users", str(userland_user_id), "auth_factors"),
            model=AuthenticationFactor,
            params=params,
            request_options=request_options,
        )

    def create_user_auth_factor(
        self,
        userland_user_id: str,
        *,
        type: Literal["totp"],
        totp_issuer: Optional[str] = None,
        totp_user: Optional[str] = None,
        totp_secret: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> UserAuthenticationFactorEnrollResponse:
        """Enroll an authentication factor

        Enrolls a user in a new [authentication factor](https://workos.com/docs/reference/authkit/mfa/authentication-factor).

        Args:
            userland_user_id: The ID of the [user](https://workos.com/docs/reference/authkit/user).
            type: The type of the factor to enroll.
            totp_issuer: Your application or company name displayed in the user's authenticator app.
            totp_user: The user's account name displayed in their authenticator app.
            totp_secret: The Base32-encoded shared secret for TOTP factors. This can be provided when creating the auth factor, otherwise it will be generated. The algorithm used to derive TOTP codes is SHA-1, the code length is 6 digits, and the timestep is 30 seconds – the secret must be compatible with these parameters.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            UserAuthenticationFactorEnrollResponse

        Raises:
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "type": type,
                "totp_issuer": totp_issuer,
                "totp_user": totp_user,
                "totp_secret": totp_secret,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="post",
            path=("user_management", "users", str(userland_user_id), "auth_factors"),
            body=body,
            model=UserAuthenticationFactorEnrollResponse,
            request_options=request_options,
        )


class AsyncMultiFactorAuth:
    """Multi Factor Auth API resources (async)."""

    def __init__(self, client: "AsyncWorkOSClient") -> None:
        self._client = client

    async def verify_challenge(
        self,
        id: str,
        *,
        code: str,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationChallengeVerifyResponse:
        """Verify Challenge

        Verifies an Authentication Challenge.

        Args:
            id: The unique ID of the Authentication Challenge.
            code: The one-time code to verify.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationChallengeVerifyResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            "code": code,
        }
        return await self._client.request(
            method="post",
            path=("auth", "challenges", str(id), "verify"),
            body=body,
            model=AuthenticationChallengeVerifyResponse,
            request_options=request_options,
        )

    async def enroll_factor(
        self,
        *,
        type: Union[AuthenticationFactorsCreateRequestType, str],
        phone_number: Optional[str] = None,
        totp_issuer: Optional[str] = None,
        totp_user: Optional[str] = None,
        user_id: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationFactorEnrolled:
        """Enroll Factor

        Enrolls an Authentication Factor to be used as an additional factor of authentication. The returned ID should be used to create an authentication Challenge.

        Args:
            type: The type of factor to enroll.
            phone_number: Required when type is 'sms'.
            totp_issuer: Required when type is 'totp'.
            totp_user: Required when type is 'totp'.
            user_id: The ID of the user to associate the factor with.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationFactorEnrolled

        Raises:
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "type": enum_value(type),
                "phone_number": phone_number,
                "totp_issuer": totp_issuer,
                "totp_user": totp_user,
                "user_id": user_id,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="post",
            path=("auth", "factors", "enroll"),
            body=body,
            model=AuthenticationFactorEnrolled,
            request_options=request_options,
        )

    async def get_factor(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationFactor:
        """Get Factor

        Gets an Authentication Factor.

        Args:
            id: The unique ID of the Factor.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationFactor

        Raises:
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return await self._client.request(
            method="get",
            path=("auth", "factors", str(id)),
            model=AuthenticationFactor,
            request_options=request_options,
        )

    async def delete_factor(
        self,
        id: str,
        *,
        request_options: Optional[RequestOptions] = None,
    ) -> None:
        """Delete Factor

        Permanently deletes an Authentication Factor. It cannot be undone.

        Args:
            id: The unique ID of the Factor.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            NotFoundError: If the resource is not found (404).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        await self._client.request(
            method="delete",
            path=("auth", "factors", str(id)),
            request_options=request_options,
        )

    async def challenge_factor(
        self,
        id: str,
        *,
        sms_template: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> AuthenticationChallenge:
        """Challenge Factor

        Creates a Challenge for an Authentication Factor.

        Args:
            id: The unique ID of the Authentication Factor to be challenged.
            sms_template: A custom template for the SMS message. Use the {{code}} placeholder to include the verification code.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AuthenticationChallenge

        Raises:
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "sms_template": sms_template,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="post",
            path=("auth", "factors", str(id), "challenge"),
            body=body,
            model=AuthenticationChallenge,
            request_options=request_options,
        )

    async def list_user_auth_factors(
        self,
        userland_user_id: str,
        *,
        limit: Optional[int] = None,
        before: Optional[str] = None,
        after: Optional[str] = None,
        order: Optional[Union[PaginationOrder, str]] = "desc",
        request_options: Optional[RequestOptions] = None,
    ) -> AsyncPage[AuthenticationFactor]:
        """List authentication factors

        Lists the [authentication factors](https://workos.com/docs/reference/authkit/mfa/authentication-factor) for a user.

        Args:
            userland_user_id: The ID of the [user](https://workos.com/docs/reference/authkit/user).
            limit: Upper limit on the number of objects to return, between `1` and `100`. Defaults to `10`.
            before: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list.
            after: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list.
            order: Order the results by the creation time. Defaults to `desc`.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AsyncPage[AuthenticationFactor]

        Raises:
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
            }.items()
            if v is not None
        }
        return await self._client.request_page(
            method="get",
            path=("user_management", "users", str(userland_user_id), "auth_factors"),
            model=AuthenticationFactor,
            params=params,
            request_options=request_options,
        )

    async def create_user_auth_factor(
        self,
        userland_user_id: str,
        *,
        type: Literal["totp"],
        totp_issuer: Optional[str] = None,
        totp_user: Optional[str] = None,
        totp_secret: Optional[str] = None,
        request_options: Optional[RequestOptions] = None,
    ) -> UserAuthenticationFactorEnrollResponse:
        """Enroll an authentication factor

        Enrolls a user in a new [authentication factor](https://workos.com/docs/reference/authkit/mfa/authentication-factor).

        Args:
            userland_user_id: The ID of the [user](https://workos.com/docs/reference/authkit/user).
            type: The type of the factor to enroll.
            totp_issuer: Your application or company name displayed in the user's authenticator app.
            totp_user: The user's account name displayed in their authenticator app.
            totp_secret: The Base32-encoded shared secret for TOTP factors. This can be provided when creating the auth factor, otherwise it will be generated. The algorithm used to derive TOTP codes is SHA-1, the code length is 6 digits, and the timestep is 30 seconds – the secret must be compatible with these parameters.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            UserAuthenticationFactorEnrollResponse

        Raises:
            UnprocessableEntityError: If the request data is unprocessable (422).
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: Dict[str, Any] = {
            k: v
            for k, v in {
                "type": type,
                "totp_issuer": totp_issuer,
                "totp_user": totp_user,
                "totp_secret": totp_secret,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="post",
            path=("user_management", "users", str(userland_user_id), "auth_factors"),
            body=body,
            model=UserAuthenticationFactorEnrollResponse,
            request_options=request_options,
        )
