mirror of
https://github.com/fastapi-users/fastapi-users.git
synced 2025-08-15 11:11:16 +08:00
136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
from typing import Generic, Sequence, Type
|
|
|
|
from fastapi import APIRouter
|
|
|
|
from fastapi_users import models, schemas
|
|
from fastapi_users.authentication import AuthenticationBackend, Authenticator
|
|
from fastapi_users.jwt import SecretType
|
|
from fastapi_users.manager import UserManagerDependency
|
|
from fastapi_users.router import (
|
|
get_auth_router,
|
|
get_register_router,
|
|
get_reset_password_router,
|
|
get_users_router,
|
|
get_verify_router,
|
|
)
|
|
|
|
try:
|
|
from httpx_oauth.oauth2 import BaseOAuth2
|
|
|
|
from fastapi_users.router import get_oauth_router
|
|
except ModuleNotFoundError: # pragma: no cover
|
|
BaseOAuth2 = Type # type: ignore
|
|
|
|
|
|
class FastAPIUsers(Generic[models.UP, schemas.U, schemas.UC, schemas.UU]):
|
|
"""
|
|
Main object that ties together the component for users authentication.
|
|
|
|
:param get_user_manager: Dependency callable getter to inject the
|
|
user manager class instance.
|
|
:param auth_backends: List of authentication backends.
|
|
:param user_schema: Pydantic schema of a public user.
|
|
:param user_create_schema: Pydantic schema for creating a user.
|
|
:param user_update_schema: Pydantic schema for updating a user.
|
|
|
|
:attribute current_user: Dependency callable getter to inject authenticated user
|
|
with a specific set of parameters.
|
|
"""
|
|
|
|
authenticator: Authenticator
|
|
_user_schema: Type[schemas.U]
|
|
_user_create_schema: Type[schemas.UC]
|
|
_user_update_schema: Type[schemas.UU]
|
|
|
|
def __init__(
|
|
self,
|
|
get_user_manager: UserManagerDependency[models.UP],
|
|
auth_backends: Sequence[AuthenticationBackend],
|
|
user_schema: Type[schemas.U],
|
|
user_create_schema: Type[schemas.UC],
|
|
user_update_schema: Type[schemas.UU],
|
|
):
|
|
self.authenticator = Authenticator(auth_backends, get_user_manager)
|
|
|
|
self._user_schema = user_schema
|
|
self._user_create_schema = user_create_schema
|
|
self._user_update_schema = user_update_schema
|
|
|
|
self.get_user_manager = get_user_manager
|
|
self.current_user = self.authenticator.current_user
|
|
|
|
def get_register_router(self) -> APIRouter:
|
|
"""Return a router with a register route."""
|
|
return get_register_router(
|
|
self.get_user_manager,
|
|
self._user_schema,
|
|
self._user_create_schema,
|
|
)
|
|
|
|
def get_verify_router(self) -> APIRouter:
|
|
"""Return a router with e-mail verification routes."""
|
|
return get_verify_router(self.get_user_manager, self._user_schema)
|
|
|
|
def get_reset_password_router(self) -> APIRouter:
|
|
"""Return a reset password process router."""
|
|
return get_reset_password_router(self.get_user_manager)
|
|
|
|
def get_auth_router(
|
|
self, backend: AuthenticationBackend, requires_verification: bool = False
|
|
) -> APIRouter:
|
|
"""
|
|
Return an auth router for a given authentication backend.
|
|
|
|
:param backend: The authentication backend instance.
|
|
:param requires_verification: Whether the authentication
|
|
require the user to be verified or not.
|
|
"""
|
|
return get_auth_router(
|
|
backend,
|
|
self.get_user_manager,
|
|
self.authenticator,
|
|
requires_verification,
|
|
)
|
|
|
|
def get_oauth_router(
|
|
self,
|
|
oauth_client: BaseOAuth2,
|
|
backend: AuthenticationBackend,
|
|
state_secret: SecretType,
|
|
redirect_url: str = None,
|
|
) -> APIRouter:
|
|
"""
|
|
Return an OAuth router for a given OAuth client and authentication backend.
|
|
|
|
:param oauth_client: The HTTPX OAuth client instance.
|
|
:param backend: The authentication backend instance.
|
|
:param state_secret: Secret used to encode the state JWT.
|
|
:param redirect_url: Optional arbitrary redirect URL for the OAuth2 flow.
|
|
If not given, the URL to the callback endpoint will be generated.
|
|
"""
|
|
return get_oauth_router(
|
|
oauth_client,
|
|
backend,
|
|
self.get_user_manager,
|
|
state_secret,
|
|
redirect_url,
|
|
)
|
|
|
|
def get_users_router(
|
|
self,
|
|
requires_verification: bool = False,
|
|
) -> APIRouter:
|
|
"""
|
|
Return a router with routes to manage users.
|
|
|
|
:param requires_verification: Whether the endpoints
|
|
require the users to be verified or not.
|
|
"""
|
|
return get_users_router(
|
|
self.get_user_manager,
|
|
self._user_schema,
|
|
self._user_update_schema,
|
|
self.authenticator,
|
|
requires_verification,
|
|
)
|